feat(operit): 增强转译层并接入 5 个 Operit 市场插件
operit_loader.py: - OkHttp 新增 newClient()（真实市场插件普遍使用，而非 newBuilder） - OkHttpRequest.execute() 改为 async，与 Operit 运行时异步语义对齐 - OkHttpResponse 新增 json() 与 statusMessage()（基于 http.HTTPStatus） operit_translate.py（新增）: - Operit .ts 插件 -> xiaoli-cli .py 脚手架生成器 - 自动提取 METADATA、生成 exports 胶水，原 TS 作参考注释保留 plugins/operit_scripts/ 新增 5 个已跑通的真实插件: - market_time 时间查询/格式化 - market_duckduckgo 网页内容抓取与清洗 - market_crossref Crossref 学术文献检索 - market_network_test HTTP 连通性测试 - market_reader 本地代码递归搜索 注: 工具需注册为 async 包装函数，同步 lambda 会导致 complete() 不执行的假阳性
This commit is contained in:
shuiyu67 committed
1 parent
f7f7033e57
commit
388ac493af
7 files changed
+868
-1
No files matched your search
@@ -26,6 +26,7 @@ import time
|
||||
import shutil
|
||||
import base64
|
||||
import hashlib
|
||||
import http
|
||||
import platform
|
||||
import subprocess
|
||||
import tempfile
|
||||
@@ -145,10 +146,18 @@ class OkHttpResponse:
|
||||
self.statusCode = status_code
|
||||
self.content = content
|
||||
self._headers = headers or {}
|
||||
try:
|
||||
self.statusMessage = http.HTTPStatus(status_code).phrase
|
||||
except Exception:
|
||||
self.statusMessage = ""
|
||||
|
||||
def isSuccessful(self) -> bool:
|
||||
return 200 <= self.statusCode < 300
|
||||
|
||||
def json(self):
|
||||
import json as _json
|
||||
return _json.loads(self.content)
|
||||
|
||||
def headers(self) -> dict:
|
||||
return self._headers
|
||||
|
||||
@@ -185,7 +194,7 @@ class OkHttpRequest:
|
||||
def build(self) -> "OkHttpRequest":
|
||||
return self
|
||||
|
||||
def execute(self) -> OkHttpResponse:
|
||||
async def execute(self) -> OkHttpResponse:
|
||||
if requests is None and httpx is None:
|
||||
raise RuntimeError("需要安装 requests 或 httpx: pip install requests")
|
||||
lib = requests or httpx
|
||||
@@ -231,6 +240,11 @@ class _OkHttp:
|
||||
def newBuilder() -> OkHttpBuilder:
|
||||
return OkHttpBuilder()
|
||||
|
||||
@staticmethod
|
||||
def newClient() -> OkHttpClient:
|
||||
# 真实 Operit 插件用 OkHttp.newClient() 而非 newBuilder().build()
|
||||
return OkHttpClient()
|
||||
|
||||
|
||||
OkHttp = _OkHttp()
|
||||
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
# METADATA
|
||||
# {
|
||||
# "name": "crossref",
|
||||
# "display_name": { "zh": "Crossref 学术文献查询", "en": "Crossref Academic Literature Search" },
|
||||
# "description": { "zh": "Crossref 学术文献查询（DOI/关键词/作者/标题/ISSN）", "en": "Crossref scholarly literature search." },
|
||||
# "category": "Search",
|
||||
# "tools": [
|
||||
# { "name": "search_by_doi", "description": { "zh": "通过 DOI 查询文章", "en": "Query by DOI" } },
|
||||
# { "name": "search_by_keyword", "description": { "zh": "通过关键词搜索", "en": "Search by keyword" } },
|
||||
# { "name": "search_by_author", "description": { "zh": "通过作者搜索", "en": "Search by author" } },
|
||||
# { "name": "search_by_title", "description": { "zh": "通过标题搜索", "en": "Search by title" } },
|
||||
# { "name": "search_by_issn", "description": { "zh": "通过 ISSN 查询", "en": "Search by ISSN" } }
|
||||
# ]
|
||||
# }
|
||||
import json as _json
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
HEADERS = {'User-Agent': 'Operit/1.0 (mailto:support@example.com)'}
|
||||
BASE_URL = "https://api.crossref.org"
|
||||
DEFAULT_ROWS = 10
|
||||
MAX_ROWS = 100
|
||||
|
||||
|
||||
def getErrorMessage(error):
|
||||
return str(error) if error else "unknown"
|
||||
|
||||
|
||||
def getErrorStack(error):
|
||||
import traceback
|
||||
return traceback.format_exc()
|
||||
|
||||
|
||||
def format_authors(authors):
|
||||
if not authors:
|
||||
return "N/A"
|
||||
return ", ".join(
|
||||
f"{a.get('given', '')} {a.get('family', '')}".strip()
|
||||
for a in authors[:5] if a.get('given') or a.get('family')
|
||||
) or "N/A"
|
||||
|
||||
|
||||
def format_date(date_parts):
|
||||
if not date_parts or not date_parts[0]:
|
||||
return "N/A"
|
||||
p = date_parts[0]
|
||||
if len(p) == 1:
|
||||
return str(p[0])
|
||||
if len(p) == 2:
|
||||
return f"{p[0]}-{str(p[1]).zfill(2)}"
|
||||
if len(p) == 3:
|
||||
return f"{p[0]}-{str(p[1]).zfill(2)}-{str(p[2]).zfill(2)}"
|
||||
return "N/A"
|
||||
|
||||
|
||||
def format_article(item, index=None):
|
||||
lines = []
|
||||
if index is not None:
|
||||
lines.append(f"\n=== Article {index + 1} ===")
|
||||
title = item.get("title", ["No Title"])[0] if item.get("title") else "No Title"
|
||||
lines.append(f"Title: {title}")
|
||||
if item.get("DOI"):
|
||||
lines.append(f"DOI: {item['DOI']}")
|
||||
lines.append(f"URL: https://doi.org/{item['DOI']}")
|
||||
lines.append(f"Authors: {format_authors(item.get('author'))}")
|
||||
dp = (item.get("published") or item.get("published-print") or item.get("published-online") or {}).get("date-parts")
|
||||
lines.append(f"Published: {format_date(dp)}")
|
||||
if item.get("container-title"):
|
||||
lines.append(f"Journal/Conference: {item['container-title'][0]}")
|
||||
if item.get("ISSN"):
|
||||
lines.append(f"ISSN: {', '.join(item['ISSN'])}")
|
||||
if item.get("publisher"):
|
||||
lines.append(f"Publisher: {item['publisher']}")
|
||||
if item.get("type"):
|
||||
lines.append(f"Type: {item['type']}")
|
||||
if item.get("is-referenced-by-count") is not None:
|
||||
lines.append(f"Citations: {item['is-referenced-by-count']}")
|
||||
if item.get("abstract"):
|
||||
ab = __import__("re").sub(r"<[^>]*>", "", item["abstract"])
|
||||
lines.append(f"Abstract: {ab[:500]}{'...' if len(ab) > 500 else ''}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
async def _get_json(url):
|
||||
client = OkHttp.newClient()
|
||||
resp = await client.newRequest().url(url).method("GET").headers(HEADERS).build().execute()
|
||||
if not resp.isSuccessful():
|
||||
raise RuntimeError(f"HTTP {resp.statusCode} - {resp.statusMessage}")
|
||||
return resp.json()
|
||||
|
||||
|
||||
async def search_by_doi(params):
|
||||
doi = (params.get("doi") or "").strip()
|
||||
if not doi:
|
||||
return {"success": False, "message": "请提供有效的 DOI"}
|
||||
try:
|
||||
data = await _get_json(f"{BASE_URL}/works/{quote_plus(doi)}")
|
||||
if data.get("status") == "ok" and data.get("message"):
|
||||
return {"success": True, "message": "查询成功", "data": format_article(data["message"])}
|
||||
return {"success": False, "message": "未找到该 DOI 对应的文章"}
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"查询失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def search_by_keyword(params):
|
||||
query = (params.get("query") or "").strip()
|
||||
if not query:
|
||||
return {"success": False, "message": "请提供有效的搜索关键词"}
|
||||
rows = max(1, min(int(params.get("rows", DEFAULT_ROWS) or DEFAULT_ROWS), MAX_ROWS))
|
||||
sort = params.get("sort", "relevance")
|
||||
order = params.get("order", "desc")
|
||||
qs = "&".join(f"{k}={quote_plus(v)}" for k, v in
|
||||
{"query": query, "rows": str(rows), "sort": sort, "order": order}.items())
|
||||
try:
|
||||
data = await _get_json(f"{BASE_URL}/works?{qs}")
|
||||
return _summarize(data, query, f'关键词 "{query}"')
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"搜索失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def search_by_author(params):
|
||||
author = (params.get("author") or "").strip()
|
||||
if not author:
|
||||
return {"success": False, "message": "请提供有效的作者名字"}
|
||||
rows = max(1, min(int(params.get("rows", DEFAULT_ROWS) or DEFAULT_ROWS), MAX_ROWS))
|
||||
qs = "&".join(f"{k}={quote_plus(v)}" for k, v in
|
||||
{"query.author": author, "rows": str(rows)}.items())
|
||||
try:
|
||||
data = await _get_json(f"{BASE_URL}/works?{qs}")
|
||||
return _summarize(data, author, f'作者 "{author}"')
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"搜索失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def search_by_title(params):
|
||||
title = (params.get("title") or "").strip()
|
||||
if not title:
|
||||
return {"success": False, "message": "请提供有效的文章标题"}
|
||||
rows = max(1, min(int(params.get("rows", DEFAULT_ROWS) or DEFAULT_ROWS), MAX_ROWS))
|
||||
qs = "&".join(f"{k}={quote_plus(v)}" for k, v in
|
||||
{"query.title": title, "rows": str(rows)}.items())
|
||||
try:
|
||||
data = await _get_json(f"{BASE_URL}/works?{qs}")
|
||||
return _summarize(data, title, f'标题 "{title}"')
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"搜索失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def search_by_issn(params):
|
||||
issn = (params.get("issn") or "").strip()
|
||||
if not issn:
|
||||
return {"success": False, "message": "请提供有效的 ISSN"}
|
||||
rows = max(1, min(int(params.get("rows", DEFAULT_ROWS) or DEFAULT_ROWS), MAX_ROWS))
|
||||
try:
|
||||
data = await _get_json(f"{BASE_URL}/journals/{quote_plus(issn)}/works?rows={rows}")
|
||||
return _summarize(data, issn, f'ISSN {issn}')
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"查询失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
def _summarize(data, key, label):
|
||||
if data.get("status") == "ok" and data.get("message") and data.get("message", {}).get("items"):
|
||||
items = data["message"]["items"]
|
||||
total = data["message"].get("total-results")
|
||||
results = [format_article(it, i) for i, it in enumerate(items)]
|
||||
return {"success": True, "message": "搜索成功",
|
||||
"data": f"Found {total} results for {label} (showing {len(items)}):\n\n" + "\n\n".join(results),
|
||||
"total": total, "count": len(items)}
|
||||
return {"success": False, "message": f"未找到匹配 {label} 的文章"}
|
||||
|
||||
|
||||
async def wrap(func, params, ok_msg="操作成功", err_msg="操作失败"):
|
||||
try:
|
||||
result = await func(params)
|
||||
complete({"success": True, "message": result.get("message", ok_msg), "data": result.get("data"),
|
||||
**{k: v for k, v in result.items() if k not in ("success", "message", "data")}})
|
||||
except Exception as error:
|
||||
console.error(f"Error: {getErrorMessage(error)}")
|
||||
complete({"success": False, "message": getErrorMessage(error), "error_stack": getErrorStack(error)})
|
||||
|
||||
|
||||
async def search_by_doi_wrapper(params):
|
||||
await wrap(search_by_doi, params, "查询成功", "查询失败")
|
||||
|
||||
|
||||
async def search_by_keyword_wrapper(params):
|
||||
await wrap(search_by_keyword, params, "搜索成功", "搜索失败")
|
||||
|
||||
|
||||
async def search_by_author_wrapper(params):
|
||||
await wrap(search_by_author, params, "搜索成功", "搜索失败")
|
||||
|
||||
|
||||
async def search_by_title_wrapper(params):
|
||||
await wrap(search_by_title, params, "搜索成功", "搜索失败")
|
||||
|
||||
|
||||
async def search_by_issn_wrapper(params):
|
||||
await wrap(search_by_issn, params, "查询成功", "查询失败")
|
||||
|
||||
|
||||
exports.search_by_doi = search_by_doi_wrapper
|
||||
exports.search_by_keyword = search_by_keyword_wrapper
|
||||
exports.search_by_author = search_by_author_wrapper
|
||||
exports.search_by_title = search_by_title_wrapper
|
||||
exports.search_by_issn = search_by_issn_wrapper
|
||||
@@ -0,0 +1,195 @@
|
||||
# METADATA
|
||||
# {
|
||||
# "name": "market_duckduckgo",
|
||||
# "display_name": {
|
||||
# "zh": "DuckDuckGo 搜索(市场插件翻译)",
|
||||
# "en": "DuckDuckGo Search (translated from operit market)"
|
||||
# },
|
||||
# "description": {
|
||||
# "zh": "从 operit 市场 duckduckgo.ts 忠实翻译：搜索与网页内容抓取。",
|
||||
# "en": "Faithful translation of operit market duckduckgo.ts: search & fetch."
|
||||
# },
|
||||
# "category": "Search",
|
||||
# "tools": [
|
||||
# {
|
||||
# "name": "search",
|
||||
# "description": {
|
||||
# "zh": "搜索 DuckDuckGo 并返回格式化结果",
|
||||
# "en": "Search DuckDuckGo and return formatted results"
|
||||
# },
|
||||
# "parameters": [
|
||||
# {"name": "query", "type": "string", "required": true},
|
||||
# {"name": "max_results", "type": "string", "required": false}
|
||||
# ]
|
||||
# },
|
||||
# {
|
||||
# "name": "fetch_content",
|
||||
# "description": {
|
||||
# "zh": "抓取并解析网页内容为纯文本",
|
||||
# "en": "Fetch and parse webpage content to plain text"
|
||||
# },
|
||||
# "parameters": [
|
||||
# {"name": "url", "type": "string", "required": true}
|
||||
# ]
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
|
||||
import re
|
||||
import html
|
||||
import asyncio
|
||||
|
||||
|
||||
BASE_URL = "https://html.duckduckgo.com/html/"
|
||||
|
||||
_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""轻量异步令牌桶，忠实于 TS 版 RateLimiter.acquire()"""
|
||||
def __init__(self, max_concurrent: int):
|
||||
self._sem = asyncio.Semaphore(max_concurrent)
|
||||
|
||||
async def acquire(self):
|
||||
await self._sem.acquire()
|
||||
|
||||
def release(self):
|
||||
self._sem.release()
|
||||
|
||||
|
||||
_search_limiter = RateLimiter(20)
|
||||
_fetch_limiter = RateLimiter(20)
|
||||
|
||||
|
||||
def getErrorMessage(error):
|
||||
return str(error) if error else "unknown"
|
||||
|
||||
|
||||
def getErrorStack(error):
|
||||
import traceback
|
||||
return traceback.format_exc()
|
||||
|
||||
|
||||
def _decode_entities(text: str) -> str:
|
||||
return html.unescape(text)
|
||||
|
||||
|
||||
def _format_results_for_llm(results: list) -> str:
|
||||
lines = []
|
||||
for r in results:
|
||||
lines.append(f"{r['position']}. {r['title']}")
|
||||
lines.append(f" URL: {r['link']}")
|
||||
lines.append(f" {r['snippet']}")
|
||||
lines.append("")
|
||||
return "\n".join(lines).strip()
|
||||
|
||||
|
||||
async def wrap(func, params, ok_msg="操作成功", err_msg="操作失败"):
|
||||
try:
|
||||
result = await func(params)
|
||||
complete({"success": True, "message": ok_msg, "data": result})
|
||||
except Exception as error:
|
||||
console.error(f"Error: {getErrorMessage(error)}")
|
||||
complete({"success": False, "message": getErrorMessage(error), "error_stack": getErrorStack(error)})
|
||||
|
||||
|
||||
async def search(params):
|
||||
"""忠实翻译 duckduckgo.ts search()"""
|
||||
query = params.get("query", "")
|
||||
max_results = 10
|
||||
if params.get("max_results"):
|
||||
try:
|
||||
parsed = int(params["max_results"])
|
||||
if not str(parsed).startswith("NaN"):
|
||||
max_results = parsed
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if not query:
|
||||
raise ValueError("查询不能为空")
|
||||
|
||||
await _search_limiter.acquire()
|
||||
try:
|
||||
console.log(f"正在从DuckDuckGo搜索: {query}")
|
||||
client = OkHttp.newClient()
|
||||
request = (client.newRequest()
|
||||
.url(BASE_URL)
|
||||
.method("POST")
|
||||
.body({"q": query, "b": "", "kl": ""}, "form")
|
||||
.headers({"User-Agent": _UA}))
|
||||
|
||||
response = await request.build().execute()
|
||||
if not response.isSuccessful():
|
||||
raise RuntimeError(f"HTTP 错误! 状态码: {response.statusCode}")
|
||||
|
||||
html_text = response.content
|
||||
results = []
|
||||
result_regex = re.compile(
|
||||
r'<h2 class="result__title">[\s\S]*?<a.*?href="([^"]+)".*?>([\s\S]+?)</a>'
|
||||
r'[\s\S]*?</h2>[\s\S]*?<a class="result__snippet".*?>([\s\S]+?)</a>',
|
||||
re.IGNORECASE,
|
||||
)
|
||||
for m in result_regex.finditer(html_text):
|
||||
if len(results) >= max_results:
|
||||
break
|
||||
link = m.group(1)
|
||||
if "y.js" in link:
|
||||
continue # 跳过广告
|
||||
if link.startswith("//duckduckgo.com/l/?uddg="):
|
||||
try:
|
||||
link = _decode_entities(link.split("uddg=")[1].split("&")[0])
|
||||
except Exception:
|
||||
console.error(f"URL 解码失败: {link}")
|
||||
title = _decode_entities(re.sub(r"<[^>]*>", "", m.group(2)).strip())
|
||||
snippet = _decode_entities(re.sub(r"<[^>]*>", "", m.group(3)).strip())
|
||||
results.append({"title": title, "link": link, "snippet": snippet,
|
||||
"position": len(results) + 1})
|
||||
|
||||
console.log(f"成功找到 {len(results)} 个结果")
|
||||
return _format_results_for_llm(results)
|
||||
finally:
|
||||
_search_limiter.release()
|
||||
|
||||
|
||||
async def fetch_content(params):
|
||||
"""忠实翻译 duckduckgo.ts fetch_content()"""
|
||||
url = params.get("url", "")
|
||||
if not url:
|
||||
raise ValueError("URL不能为空")
|
||||
|
||||
await _fetch_limiter.acquire()
|
||||
try:
|
||||
console.log(f"正在抓取内容: {url}")
|
||||
client = OkHttp.newClient()
|
||||
request = (client.newRequest()
|
||||
.url(url)
|
||||
.method("GET")
|
||||
.headers({"User-Agent": _UA}))
|
||||
|
||||
response = await request.build().execute()
|
||||
if not response.isSuccessful():
|
||||
raise RuntimeError(f"无法访问网页 ({response.statusCode})")
|
||||
|
||||
text = response.content
|
||||
for tag in ("style", "script", "nav", "header", "footer"):
|
||||
text = re.sub(rf"<{tag}[^>]*>[\s\S]*?</{tag}>", "", text, flags=re.IGNORECASE)
|
||||
text = re.sub(r"<[^>]+>", " ", text)
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
if len(text) > 8000:
|
||||
text = text[:8000] + "... [内容被截断]"
|
||||
console.log(f"成功抓取并解析内容 ({len(text)} 字符)")
|
||||
return text
|
||||
finally:
|
||||
_fetch_limiter.release()
|
||||
|
||||
|
||||
async def search_wrapper(params):
|
||||
await wrap(search, params, "搜索完成", "搜索失败")
|
||||
|
||||
|
||||
async def fetch_content_wrapper(params):
|
||||
await wrap(fetch_content, params, "抓取完成", "抓取失败")
|
||||
|
||||
|
||||
exports.search = search_wrapper
|
||||
exports.fetch_content = fetch_content_wrapper
|
||||
@@ -0,0 +1,90 @@
|
||||
# METADATA
|
||||
# {
|
||||
# "name": "network_test",
|
||||
# "display_name": { "zh": "网络测试", "en": "Network Test" },
|
||||
# "description": { "zh": "基于 OkHttp 的网络请求测试（GET/POST）", "en": "Network request testing via OkHttp." },
|
||||
# "category": "Network",
|
||||
# "tools": [
|
||||
# { "name": "http_get", "description": { "zh": "发送 HTTP GET 请求", "en": "Send HTTP GET" } },
|
||||
# { "name": "http_post", "description": { "zh": "发送 HTTP POST 请求", "en": "Send HTTP POST" } }
|
||||
# ]
|
||||
# }
|
||||
import json as _json
|
||||
|
||||
|
||||
def getErrorMessage(error):
|
||||
return str(error) if error else "unknown"
|
||||
|
||||
|
||||
def getErrorStack(error):
|
||||
import traceback
|
||||
return traceback.format_exc()
|
||||
|
||||
|
||||
def format_response(resp):
|
||||
content = resp.content or ""
|
||||
result = {
|
||||
"status_code": resp.statusCode,
|
||||
"status_message": resp.statusMessage,
|
||||
"content_length": len(content),
|
||||
"headers": resp.headers(),
|
||||
"content": content[:12000],
|
||||
}
|
||||
try:
|
||||
result["json"] = _json.loads(content)
|
||||
except Exception:
|
||||
result["json"] = None
|
||||
return result
|
||||
|
||||
|
||||
async def http_get(params):
|
||||
if not params.get("url"):
|
||||
return {"success": False, "message": "URL不能为空"}
|
||||
try:
|
||||
console.log(f"发送GET请求: {params['url']}")
|
||||
req = OkHttp.newClient().newRequest().url(params["url"]).method("GET")
|
||||
if params.get("headers"):
|
||||
req = req.headers(params["headers"])
|
||||
resp = await req.build().execute()
|
||||
return {"success": True, "message": "GET请求完成", "data": format_response(resp)}
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"GET请求失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def http_post(params):
|
||||
if not params.get("url"):
|
||||
return {"success": False, "message": "URL不能为空"}
|
||||
if params.get("body") is None:
|
||||
return {"success": False, "message": "请求体不能为空"}
|
||||
try:
|
||||
req = OkHttp.newClient().newRequest().url(params["url"]).method("POST")
|
||||
if params.get("headers"):
|
||||
req = req.headers(params["headers"])
|
||||
body_type = params.get("body_type", "json")
|
||||
req = req.body(params["body"], body_type)
|
||||
resp = await req.build().execute()
|
||||
return {"success": True, "message": "POST请求完成", "data": format_response(resp)}
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"POST请求失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def wrap(func, params, ok_msg="操作成功", err_msg="操作失败"):
|
||||
try:
|
||||
result = await func(params)
|
||||
complete({"success": True, "message": result.get("message", ok_msg),
|
||||
"data": result.get("data"), **{k: v for k, v in result.items() if k not in ("success", "message", "data")}})
|
||||
except Exception as error:
|
||||
console.error(f"Error: {getErrorMessage(error)}")
|
||||
complete({"success": False, "message": getErrorMessage(error), "error_stack": getErrorStack(error)})
|
||||
|
||||
|
||||
async def http_get_wrapper(params):
|
||||
await wrap(http_get, params, "GET请求完成", "GET请求失败")
|
||||
|
||||
|
||||
async def http_post_wrapper(params):
|
||||
await wrap(http_post, params, "POST请求完成", "POST请求失败")
|
||||
|
||||
|
||||
exports.http_get = http_get_wrapper
|
||||
exports.http_post = http_post_wrapper
|
||||
@@ -0,0 +1,85 @@
|
||||
# METADATA
|
||||
# {
|
||||
# "name": "reader",
|
||||
# "display_name": { "zh": "代码读取器", "en": "Code Reader" },
|
||||
# "description": { "zh": "在文件夹中搜索匹配的代码（本地文件，无需 Android）", "en": "Search matching code in a folder." },
|
||||
# "category": "File",
|
||||
# "tools": [
|
||||
# { "name": "search_code_in_folder", "description": { "zh": "在文件夹中搜索匹配的代码", "en": "Search matching code in a folder" } }
|
||||
# ]
|
||||
# }
|
||||
import os
|
||||
import re
|
||||
|
||||
|
||||
def getErrorMessage(error):
|
||||
return str(error) if error else "unknown"
|
||||
|
||||
|
||||
def getErrorStack(error):
|
||||
import traceback
|
||||
return traceback.format_exc()
|
||||
|
||||
|
||||
async def search_code_in_folder(params):
|
||||
folder_path = params.get("folder_path")
|
||||
pattern = params.get("pattern")
|
||||
file_extensions = params.get("file_extensions", ".js,.ts,.jsx,.tsx,.java,.cs,.py")
|
||||
recursive = str(params.get("recursive", "true")).lower() in ("true", "1", "yes")
|
||||
if not folder_path:
|
||||
return {"success": False, "message": "文件夹路径不能为空"}
|
||||
if not pattern:
|
||||
return {"success": False, "message": "搜索模式不能为空"}
|
||||
try:
|
||||
exts = [e.strip().lower() for e in file_extensions.split(",") if e.strip()]
|
||||
rx = re.compile(pattern, re.MULTILINE)
|
||||
files = []
|
||||
if recursive:
|
||||
for root, _, fnames in os.walk(folder_path):
|
||||
for fn in fnames:
|
||||
if any(fn.lower().endswith(e) for e in exts):
|
||||
files.append(os.path.join(root, fn))
|
||||
else:
|
||||
for fn in os.listdir(folder_path):
|
||||
fp = os.path.join(folder_path, fn)
|
||||
if os.path.isfile(fp) and any(fn.lower().endswith(e) for e in exts):
|
||||
files.append(fp)
|
||||
|
||||
results = []
|
||||
for fp in files:
|
||||
try:
|
||||
with open(fp, "r", encoding="utf-8", errors="replace") as fh:
|
||||
content = fh.read()
|
||||
except Exception as e:
|
||||
console.error(f"处理文件 {fp} 时出错: {getErrorMessage(e)}")
|
||||
continue
|
||||
matches = []
|
||||
for m in rx.finditer(content):
|
||||
ln = content.count("\n", 0, m.start()) + 1
|
||||
matches.append({"match": m.group(0), "line": ln,
|
||||
"content": content.split("\n")[ln - 1] if ln - 1 < len(content.split("\n")) else ""})
|
||||
if matches:
|
||||
results.append({"file": fp, "matches": matches})
|
||||
total = sum(len(r["matches"]) for r in results)
|
||||
return {"success": True, "message": f"在 {len(files)} 个文件中找到 {total} 处匹配",
|
||||
"data": {"files_searched": len(files), "files_with_matches": len(results),
|
||||
"total_matches": total, "results": results[:50]}}
|
||||
except Exception as e:
|
||||
return {"success": False, "message": f"搜索失败: {getErrorMessage(e)}"}
|
||||
|
||||
|
||||
async def wrap(func, params, ok_msg="操作成功", err_msg="操作失败"):
|
||||
try:
|
||||
result = await func(params)
|
||||
complete({"success": True, "message": result.get("message", ok_msg),
|
||||
"data": result.get("data"), **{k: v for k, v in result.items() if k not in ("success", "message", "data")}})
|
||||
except Exception as error:
|
||||
console.error(f"Error: {getErrorMessage(error)}")
|
||||
complete({"success": False, "message": getErrorMessage(error), "error_stack": getErrorStack(error)})
|
||||
|
||||
|
||||
async def search_code_in_folder_wrapper(params):
|
||||
await wrap(search_code_in_folder, params, "代码搜索完成", "代码搜索失败")
|
||||
|
||||
|
||||
exports.search_code_in_folder = search_code_in_folder_wrapper
|
||||
@@ -0,0 +1,103 @@
|
||||
# METADATA
|
||||
# {
|
||||
# "name": "market_time",
|
||||
# "display_name": {
|
||||
# "zh": "时间(市场插件翻译)",
|
||||
# "en": "Time (translated from operit market)"
|
||||
# },
|
||||
# "description": {
|
||||
# "zh": "从 operit 市场 time.ts 忠实翻译：获取/格式化当前时间。",
|
||||
# "en": "Faithful translation of operit market time.ts: get/format current time."
|
||||
# },
|
||||
# "category": "Utility",
|
||||
# "tools": [
|
||||
# {
|
||||
# "name": "get_time",
|
||||
# "description": {
|
||||
# "zh": "获取当前时间各字段",
|
||||
# "en": "Get current time fields"
|
||||
# },
|
||||
# "parameters": []
|
||||
# },
|
||||
# {
|
||||
# "name": "format_time",
|
||||
# "description": {
|
||||
# "zh": "格式化当前时间为 24h/12h 字符串",
|
||||
# "en": "Format current time as 24h/12h strings"
|
||||
# },
|
||||
# "parameters": []
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
|
||||
import datetime
|
||||
|
||||
|
||||
def getErrorMessage(error):
|
||||
return str(error) if error else "unknown"
|
||||
|
||||
|
||||
def getErrorStack(error):
|
||||
import traceback
|
||||
return traceback.format_exc()
|
||||
|
||||
|
||||
async def wrap(func, params, ok_msg="操作成功", err_msg="操作失败"):
|
||||
try:
|
||||
result = await func(params)
|
||||
complete({"success": True, "message": ok_msg, "data": result})
|
||||
except Exception as error:
|
||||
console.error(f"Error: {getErrorMessage(error)}")
|
||||
complete({"success": False, "message": getErrorMessage(error), "error_stack": getErrorStack(error)})
|
||||
|
||||
|
||||
async def get_time(params):
|
||||
"""获取当前时间各字段（忠实于 time.ts get_time）"""
|
||||
now = datetime.datetime.now()
|
||||
return {
|
||||
"timestamp": int(now.timestamp() * 1000),
|
||||
"iso": now.isoformat(),
|
||||
"local": now.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"date": {
|
||||
"year": now.year,
|
||||
"month": now.month,
|
||||
"day": now.day,
|
||||
"weekday": now.strftime("%A"),
|
||||
},
|
||||
"time": {
|
||||
"hours": now.hour,
|
||||
"minutes": now.minute,
|
||||
"seconds": now.second,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
async def format_time(params):
|
||||
"""格式化当前时间为 24h/12h 字符串（忠实于 time.ts format_time）"""
|
||||
now = datetime.datetime.now()
|
||||
pad = lambda n: str(n).zfill(2)
|
||||
hours = now.hour
|
||||
minutes = now.minute
|
||||
seconds = now.second
|
||||
time24h = f"{pad(hours)}:{pad(minutes)}:{pad(seconds)}"
|
||||
suffix = "PM" if hours >= 12 else "AM"
|
||||
hour12 = 12 if hours % 12 == 0 else hours % 12
|
||||
time12h = f"{pad(hour12)}:{pad(minutes)}:{pad(seconds)} {suffix}"
|
||||
return {
|
||||
"iso": now.isoformat(),
|
||||
"date": f"{now.year}-{pad(now.month)}-{pad(now.day)}",
|
||||
"time24h": time24h,
|
||||
"time12h": time12h,
|
||||
}
|
||||
|
||||
|
||||
async def get_time_wrapper(params):
|
||||
await wrap(get_time, params, "时间获取完成", "时间获取失败")
|
||||
|
||||
|
||||
async def format_time_wrapper(params):
|
||||
await wrap(format_time, params, "时间格式化完成", "时间格式化失败")
|
||||
|
||||
|
||||
exports.get_time = get_time_wrapper
|
||||
exports.format_time = format_time_wrapper
|
||||
@@ -0,0 +1,175 @@
|
||||
"""operit 市场插件 (.ts) → xiaoli-cli 可运行 .py 的脚手架生成器。
|
||||
|
||||
设计原则（不需要 android_pyapi）：
|
||||
operit 插件使用的是「通用平台 API」(OkHttp / Tools / toolCall /
|
||||
NativeInterface / complete / console / exports)，不是 Android API。
|
||||
xiaoli-cli 的 operit_loader 已经 1:1 镜像了这套 API（全部用 requests/
|
||||
httpx 在 Python 端原生实现），所以「翻译」只是把 TS 语法换成 Python 语法，
|
||||
全局名原样保留 —— 不存在 Android 依赖。
|
||||
|
||||
本工具只做「可靠」的样板自动化：
|
||||
1. 从 /* METADATA {json} */ 提取插件元数据，生成为 Python 友好的 # METADATA 块
|
||||
2. 生成 run_tool 契约所需的胶水：getErrorMessage / getErrorStack / wrap /
|
||||
exports.x = wrapper
|
||||
3. 为每个工具生成 async def 骨架，并把原 .ts 函数体作为 `/// TS 参考` 注释嵌入
|
||||
4. 复杂逻辑用 TODO 标注，绝不像「全量 TS→PY 编译器」那样产生半坏代码
|
||||
|
||||
用法：
|
||||
python operit_translate.py <市场插件.ts> [-o <输出.py>] [-n <插件名>]
|
||||
"""
|
||||
import sys
|
||||
import json
|
||||
import re
|
||||
import os
|
||||
|
||||
|
||||
def extract_metadata(ts: str):
|
||||
"""从 /* METADATA ... */ 提取 JSON 元数据，返回 (meta_dict, cleaned_ts)。"""
|
||||
m = re.search(r"/\*\s*METADATA\s*(.*?)\*/", ts, re.DOTALL)
|
||||
if not m:
|
||||
return None, ts
|
||||
raw = m.group(1)
|
||||
try:
|
||||
meta = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
# 宽松：去注释后重试
|
||||
try:
|
||||
meta = json.loads(re.sub(r"//.*", "", raw))
|
||||
except json.JSONDecodeError:
|
||||
meta = None
|
||||
cleaned = ts[:m.start()] + ts[m.end():]
|
||||
return meta, cleaned
|
||||
|
||||
|
||||
def py_meta_block(meta: dict) -> str:
|
||||
"""把元数据渲染成 # METADATA 块（operit_loader 逐行解析的格式）。"""
|
||||
js = json.dumps(meta, ensure_ascii=False, indent=2)
|
||||
lines = ["# METADATA", "# {"] + ["# " + ln for ln in js[1:-1].split("\n")] + ["# }", ""]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def strip_ts_types_and_comments(ts: str) -> str:
|
||||
"""去掉类型注解与注释，仅用于把 .ts 体作为参考注释嵌入（不参与执行）。"""
|
||||
# 去块注释 / 行注释
|
||||
ts = re.sub(r"/\*.*?\*/", "", ts, flags=re.DOTALL)
|
||||
ts = re.sub(r"//[^\n]*", "", ts)
|
||||
# as 断言
|
||||
ts = re.sub(r"\bas\s+[\w\[\]]+", "", ts)
|
||||
# 泛型
|
||||
ts = re.sub(r"\b\w+<[^>]+>", lambda m: m.group(0).split("<")[0], ts)
|
||||
# 修饰符
|
||||
ts = re.sub(r"\b(private|public|protected|readonly|static|export|async|const|let|var)\b", "", ts)
|
||||
# 参数/返回类型
|
||||
ts = re.sub(r"\)\s*:\s*[^{]+?(\{)", r")\1", ts)
|
||||
ts = re.sub(r"\(([^()]*)\)\s*:\s*[^{]+", r"(\1)", ts)
|
||||
return ts
|
||||
|
||||
|
||||
def indented(block: str, prefix: str = " ") -> str:
|
||||
return "\n".join((prefix + ln if ln.strip() else ln) for ln in block.split("\n"))
|
||||
|
||||
|
||||
def generate(py_name: str, ts_path: str) -> str:
|
||||
with open(ts_path, "r", encoding="utf-8") as f:
|
||||
ts = f.read()
|
||||
|
||||
meta, cleaned = extract_metadata(ts)
|
||||
if not meta:
|
||||
meta = {"name": py_name, "display_name": {"zh": py_name, "en": py_name},
|
||||
"description": {"zh": "从 operit 市场翻译", "en": "Translated from operit market"},
|
||||
"category": "Utility", "tools": []}
|
||||
|
||||
tools = meta.get("tools", [])
|
||||
if not tools:
|
||||
# 退路：从 exports.x = 猜测工具名
|
||||
for mm in re.finditer(r"exports\.(\w+)\s*=", cleaned):
|
||||
tools.append({"name": mm.group(1)})
|
||||
|
||||
out = []
|
||||
out.append(py_meta_block(meta))
|
||||
out.append("import re")
|
||||
out.append("import html")
|
||||
out.append("import asyncio")
|
||||
out.append("")
|
||||
out.append("")
|
||||
out.append("def getErrorMessage(error):")
|
||||
out.append(" return str(error) if error else \"unknown\"")
|
||||
out.append("")
|
||||
out.append("")
|
||||
out.append("def getErrorStack(error):")
|
||||
out.append(" import traceback")
|
||||
out.append(" return traceback.format_exc()")
|
||||
out.append("")
|
||||
out.append("")
|
||||
out.append("async def wrap(func, params, ok_msg=\"操作成功\", err_msg=\"操作失败\"):")
|
||||
out.append(" \"\"\"operit 的 complete({success,message,data}) 契约封装\"\"\"")
|
||||
out.append(" try:")
|
||||
out.append(" result = await func(params)")
|
||||
out.append(" complete({\"success\": True, \"message\": ok_msg, \"data\": result})")
|
||||
out.append(" except Exception as error:")
|
||||
out.append(" console.error(f\"Error: {getErrorMessage(error)}\")")
|
||||
out.append(" complete({\"success\": False, \"message\": getErrorMessage(error),")
|
||||
out.append(" \"error_stack\": getErrorStack(error)})")
|
||||
out.append("")
|
||||
out.append("")
|
||||
# 参考注释：原 .ts 去类型后的体
|
||||
ref = strip_ts_types_and_comments(cleaned).strip()
|
||||
if ref:
|
||||
out.append("# ==================== 原 .ts 参考（去类型/注释，仅供翻译对照） ====================")
|
||||
for ln in ref.split("\n"):
|
||||
out.append("# " + ln)
|
||||
out.append("# ========================================================================================")
|
||||
out.append("")
|
||||
|
||||
# 每个工具：骨架 + wrapper + exports
|
||||
for t in tools:
|
||||
name = t.get("name", "tool")
|
||||
desc = t.get("description", {})
|
||||
desc_zh = desc.get("zh", "") if isinstance(desc, dict) else str(desc)
|
||||
out.append("")
|
||||
out.append(f"async def {name}(params):")
|
||||
out.append(f" \"\"\"{desc_zh or name}（译自 operit 市场 {meta.get('name','?')}.ts）\"\"\"")
|
||||
out.append(" # TODO: 用下面的 1:1 API 映射把 .ts 逻辑补全（参考上方 /// TS 参考）")
|
||||
out.append(" # OkHttp.newClient().newRequest().url(u).method('GET'/'POST')")
|
||||
out.append(" # .body(payload, 'json'|'form').headers({...}).build().execute() # async")
|
||||
out.append(" # response.isSuccessful() / response.statusCode / response.content")
|
||||
out.append(" # Tools.System.sleep(ms) -> {'requestedMs':, 'sleptMs':}")
|
||||
out.append(" # console.log / console.error / complete({success,message,data})")
|
||||
out.append(f' raise NotImplementedError("{name} 待翻译")')
|
||||
out.append("")
|
||||
out.append("")
|
||||
out.append(f"async def {name}_wrapper(params):")
|
||||
out.append(f" await wrap({name}, params, \"{name} 完成\", \"{name} 失败\")")
|
||||
out.append("")
|
||||
out.append("")
|
||||
out.append(f"exports.{name} = {name}_wrapper")
|
||||
out.append("")
|
||||
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print(__doc__)
|
||||
sys.exit(1)
|
||||
ts_path = sys.argv[1]
|
||||
py_name = None
|
||||
out_path = None
|
||||
if "-n" in sys.argv:
|
||||
py_name = sys.argv[sys.argv.index("-n") + 1]
|
||||
if "-o" in sys.argv:
|
||||
out_path = sys.argv[sys.argv.index("-o") + 1]
|
||||
if not py_name:
|
||||
py_name = os.path.splitext(os.path.basename(ts_path))[0].replace("-", "_")
|
||||
if not out_path:
|
||||
out_path = os.path.join(os.path.dirname(ts_path), py_name + ".py")
|
||||
|
||||
code = generate(py_name, ts_path)
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
print(f"已生成脚手架: {out_path} ({len(code)} 字节)")
|
||||
print("下一步: 打开文件，把 TODO 处按 1:1 API 映射补全即可（无需 android_pyapi）。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user