#!/usr/bin/env python3 """Fetch AMiner paper recommendations and render results as Markdown.""" from __future__ import annotations import argparse import json import pathlib import sys import urllib.error import urllib.request from typing import Any DEFAULT_SIZE = 5 MAX_SIZE = 20 REQUEST_TIMEOUT = 30 CONFIG_PATHS = [ pathlib.Path.cwd() / ".z-ai-config", pathlib.Path.home() / ".z-ai-config", pathlib.Path("/etc/.z-ai-config"), ] def load_config() -> dict[str, Any]: for path in CONFIG_PATHS: try: cfg = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: continue except (OSError, json.JSONDecodeError) as exc: raise SystemExit(f"failed to read {path}: {exc}") from exc if cfg.get("baseUrl") and cfg.get("apiKey"): if not cfg.get("token"): print( f"warning: {path} has no 'token' field; the API requires " "X-Token for auth and will return 401.", file=sys.stderr, ) return cfg raise SystemExit( "z-ai config not found. Create .z-ai-config in cwd, $HOME, or /etc " "with {\"baseUrl\": ..., \"apiKey\": ..., \"token\": ...}." ) def _clean(text: Any) -> str: return " ".join(str(text or "").split()).strip() def _truncate(text: str, limit: int) -> str: text = _clean(text) return text if len(text) <= limit else text[:limit].rstrip() + "…" def build_payload(args: argparse.Namespace) -> dict[str, Any]: arguments: dict[str, Any] = {} topics = [t for t in (args.topic or []) if _clean(t)] if topics: arguments["topics"] = topics if args.author_name: arguments["author_name"] = args.author_name if args.author_org: arguments["author_org"] = args.author_org if args.aminer_author_id: arguments["aminer_author_id"] = args.aminer_author_id if args.language_sort in {"zh", "en"}: arguments["language_sort"] = args.language_sort if args.start_year: arguments["start_year"] = args.start_year if args.end_year: arguments["end_year"] = args.end_year size = args.size if args.size else DEFAULT_SIZE size = max(1, min(size, MAX_SIZE)) arguments["size"] = size return {"function_name": "aminer_recommend", "arguments": arguments} def call_api(config: dict[str, Any], payload: dict[str, Any]) -> list[dict[str, Any]]: base_url = str(config["baseUrl"]).rstrip("/") url = f"{base_url}/functions/invoke" headers = { "Content-Type": "application/json", "User-Agent": "aminer-daily-paper-skill/1.0", "Authorization": f"Bearer {config['apiKey']}", "X-Z-AI-From": "Z", } if config.get("token"): headers["X-Token"] = str(config["token"]) if config.get("chatId"): headers["X-Chat-Id"] = str(config["chatId"]) if config.get("userId"): headers["X-User-Id"] = str(config["userId"]) body = json.dumps(payload, ensure_ascii=False).encode("utf-8") request = urllib.request.Request(url, data=body, headers=headers, method="POST") try: with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as resp: # nosec B310 raw = resp.read().decode("utf-8") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") if exc.fp else "" hint = "" if exc.code == 401 and "X-Token" in detail: hint = " (hint: add a valid 'token' field to .z-ai-config)" elif exc.code == 403: hint = " (hint: request rejected by auth — check 'apiKey' / X-Z-AI-From)" raise SystemExit(f"http {exc.code}: {detail or exc.reason}{hint}") from exc except urllib.error.URLError as exc: raise SystemExit(f"api unreachable: {exc.reason}") from exc try: parsed = json.loads(raw) except json.JSONDecodeError as exc: raise SystemExit(f"invalid json response: {exc}") from exc if isinstance(parsed.get("error"), str) and parsed["error"]: raise SystemExit(f"api error: {parsed['error']}") result = parsed.get("result") if not isinstance(result, list): raise SystemExit(f"unexpected response shape: {raw[:300]}") return [p for p in result if isinstance(p, dict)] def render_markdown(papers: list[dict[str, Any]], topics: list[str]) -> str: if not papers: return "No papers returned. Try broadening the topics or adjusting the query." lines: list[str] = [] header = f"Recommended {len(papers)} paper(s)" if topics: header += f" (topics: {' / '.join(topics[:5])})" lines.append(header) for idx, paper in enumerate(papers, start=1): lines.append("") lines.append("---") lines.append("") title = _clean(paper.get("title")) links = paper.get("links") if isinstance(paper.get("links"), dict) else {} url = _clean(paper.get("paper_url") or links.get("aminer") or links.get("arxiv")) title_line = f"**{idx}. [{title}]({url})**" if url else f"**{idx}. {title}**" lines.append(title_line) meta: list[str] = [] year = paper.get("year") if year: meta.append(f"Year: {year}") keywords = paper.get("keywords") or [] if keywords: meta.append("Keywords: " + " / ".join(str(k) for k in keywords[:5])) if meta: lines.append(" | ".join(meta)) authors = paper.get("authors") or [] if authors: author_str = ", ".join(str(a) for a in authors[:6]) if len(authors) > 6: author_str += " et al." lines.append(f"Authors: {author_str}") summary = _clean(paper.get("summary")) if summary: lines.append("") lines.append(_truncate(summary, 300)) return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description="Fetch AMiner paper recommendations.") parser.add_argument("--topic", action="append", default=[], help="research topic (repeatable)") parser.add_argument("--author-name", default="") parser.add_argument("--author-org", default="") parser.add_argument("--aminer-author-id", default="") parser.add_argument("--size", type=int, default=0) parser.add_argument("--language-sort", default="", choices=["", "zh", "en"]) parser.add_argument("--start-year", type=int, default=0) parser.add_argument("--end-year", type=int, default=0) args = parser.parse_args() payload = build_payload(args) config = load_config() papers = call_api(config, payload) topics = payload["arguments"].get("topics") or [] print(render_markdown(papers, topics)) return 0 if __name__ == "__main__": raise SystemExit(main())