diff --git a/plugins/operit_loader.py b/plugins/operit_loader.py index 32bc027..99d501e 100644 --- a/plugins/operit_loader.py +++ b/plugins/operit_loader.py @@ -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() diff --git a/plugins/operit_scripts/market_crossref.py b/plugins/operit_scripts/market_crossref.py new file mode 100644 index 0000000..56d0f7f --- /dev/null +++ b/plugins/operit_scripts/market_crossref.py @@ -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 diff --git a/plugins/operit_scripts/market_duckduckgo.py b/plugins/operit_scripts/market_duckduckgo.py new file mode 100644 index 0000000..65366e5 --- /dev/null +++ b/plugins/operit_scripts/market_duckduckgo.py @@ -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'

[\s\S]*?([\s\S]+?)' + r'[\s\S]*?

[\s\S]*?([\s\S]+?)', + 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]*?", "", 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 diff --git a/plugins/operit_scripts/market_network_test.py b/plugins/operit_scripts/market_network_test.py new file mode 100644 index 0000000..65639c1 --- /dev/null +++ b/plugins/operit_scripts/market_network_test.py @@ -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 diff --git a/plugins/operit_scripts/market_reader.py b/plugins/operit_scripts/market_reader.py new file mode 100644 index 0000000..28b12d9 --- /dev/null +++ b/plugins/operit_scripts/market_reader.py @@ -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 diff --git a/plugins/operit_scripts/market_time.py b/plugins/operit_scripts/market_time.py new file mode 100644 index 0000000..607414f --- /dev/null +++ b/plugins/operit_scripts/market_time.py @@ -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 diff --git a/plugins/operit_translate.py b/plugins/operit_translate.py new file mode 100644 index 0000000..6ec6bec --- /dev/null +++ b/plugins/operit_translate.py @@ -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()