From 788514ce8e9ffaa026302d28cc327d340ffb5292 Mon Sep 17 00:00:00 2001 From: YJLi-new Date: Thu, 5 Mar 2026 20:54:33 +0800 Subject: [PATCH] feat: add Xiaohongshu source and Reddit public fallback - add xiaohongshu/xhs source path via xiaohongshu-mcp HTTP API\n- add Reddit public JSON fallback when OpenAI auth is unavailable\n- update diagnostics/UI rendering for new source availability states\n- harden Xiaohongshu availability probe to reduce false negatives\n- include source status reporting for Xiaohongshu --- scripts/last30days.py | 180 ++++++++++++++++++++++++++++----- scripts/lib/env.py | 119 +++++++++++++--------- scripts/lib/openai_reddit.py | 99 ++++++++++++++++++ scripts/lib/render.py | 14 +++ scripts/lib/ui.py | 24 ++++- scripts/lib/xiaohongshu_api.py | 162 +++++++++++++++++++++++++++++ 6 files changed, 523 insertions(+), 75 deletions(-) create mode 100644 scripts/lib/xiaohongshu_api.py diff --git a/scripts/last30days.py b/scripts/last30days.py index 780cece..cc65356 100644 --- a/scripts/last30days.py +++ b/scripts/last30days.py @@ -44,7 +44,10 @@ TIMEOUT_PROFILES = { } # Valid source names for the --search flag -VALID_SEARCH_SOURCES = {"reddit", "x", "hn", "youtube", "tiktok", "instagram", "polymarket", "web"} +VALID_SEARCH_SOURCES = { + "reddit", "x", "hn", "youtube", "tiktok", "instagram", + "polymarket", "web", "xiaohongshu", "xhs", +} def parse_search_flag(search_str: str) -> set: @@ -64,6 +67,8 @@ def parse_search_flag(search_str: str) -> set: s = s.strip().lower() if not s: continue + if s == "xhs": + s = "xiaohongshu" if s not in VALID_SEARCH_SOURCES: print( f"Error: Unknown search source '{s}'. " @@ -133,6 +138,7 @@ from lib import ( dates, dedupe, hackernews, + xiaohongshu_api, polymarket, entity_extract, env, @@ -181,30 +187,51 @@ def _search_reddit( if mock: raw_openai = load_fixture("openai_sample.json") + reddit_items = openai_reddit.parse_reddit_response(raw_openai or {}) else: - try: - raw_openai = openai_reddit.search_reddit( - config["OPENAI_API_KEY"], - selected_models["openai"], - topic, - from_date, - to_date, - depth=depth, - auth_source=config.get("OPENAI_AUTH_SOURCE", "api_key"), - account_id=config.get("OPENAI_CHATGPT_ACCOUNT_ID"), - ) - except http.HTTPError as e: - raw_openai = {"error": str(e)} - reddit_error = f"API error: {e}" - except Exception as e: - raw_openai = {"error": str(e)} - reddit_error = f"{type(e).__name__}: {e}" + # Prefer OpenAI/Codex path when credentials are available. + if config.get("OPENAI_API_KEY"): + try: + raw_openai = openai_reddit.search_reddit( + config["OPENAI_API_KEY"], + selected_models["openai"], + topic, + from_date, + to_date, + depth=depth, + auth_source=config.get("OPENAI_AUTH_SOURCE", "api_key"), + account_id=config.get("OPENAI_CHATGPT_ACCOUNT_ID"), + ) + except http.HTTPError as e: + raw_openai = {"error": str(e)} + reddit_error = f"API error: {e}" + except Exception as e: + raw_openai = {"error": str(e)} + reddit_error = f"{type(e).__name__}: {e}" - # Parse response - reddit_items = openai_reddit.parse_reddit_response(raw_openai or {}) + # Parse response + reddit_items = openai_reddit.parse_reddit_response(raw_openai or {}) + else: + # No OpenAI auth: direct Reddit public JSON fallback. + try: + reddit_items = openai_reddit.search_reddit_public( + topic, + from_date, + to_date, + depth=depth, + ) + raw_openai = {"source": "reddit_public", "items": reddit_items} + except http.HTTPError as e: + reddit_items = [] + raw_openai = {"error": str(e), "source": "reddit_public"} + reddit_error = f"Reddit public API error: {e}" + except Exception as e: + reddit_items = [] + raw_openai = {"error": str(e), "source": "reddit_public"} + reddit_error = f"Reddit public search error: {type(e).__name__}: {e}" # Quick retry with simpler query if few results - if len(reddit_items) < 5 and not mock and not reddit_error: + if len(reddit_items) < 5 and not mock and not reddit_error and config.get("OPENAI_API_KEY"): core = openai_reddit._extract_core_subject(topic) if core.lower() != topic.lower(): try: @@ -227,7 +254,7 @@ def _search_reddit( pass # Subreddit-targeted fallback if still < 3 results - if len(reddit_items) < 3 and not mock and not reddit_error: + if len(reddit_items) < 3 and not mock and not reddit_error and config.get("OPENAI_API_KEY"): sub_query = openai_reddit._build_subreddit_query(topic) try: sub_raw = openai_reddit.search_reddit( @@ -511,6 +538,48 @@ def _search_web( return raw_results, web_error +def _search_xiaohongshu( + topic: str, + config: dict, + from_date: str, + to_date: str, + depth: str, +) -> tuple: + """Search Xiaohongshu via xiaohongshu-mcp HTTP API (runs in thread). + + Returns: + Tuple of (xiaohongshu_items, xiaohongshu_error) + Items are in web-item dict shape and can be normalized with websearch module. + """ + base_url = env.get_xiaohongshu_api_base(config) + try: + items = xiaohongshu_api.search_feeds( + topic=topic, + from_date=from_date, + to_date=to_date, + base_url=base_url, + depth=depth, + ) + except Exception as e: + return [], f"{type(e).__name__}: {e}" + + # Ensure all required keys exist for normalize_websearch_items() + for i, item in enumerate(items): + item.setdefault("id", f"XHS{i+1}") + item.setdefault("title", "") + item.setdefault("url", "") + item.setdefault("source_domain", "xiaohongshu.com") + item.setdefault("snippet", "") + if item.get("date") and not item.get("date_confidence"): + item["date_confidence"] = "med" + elif not item.get("date"): + item["date_confidence"] = "low" + item.setdefault("relevance", 0.5) + item.setdefault("why_relevant", "") + + return items, None + + def _run_supplemental( topic: str, reddit_items: list, @@ -695,6 +764,7 @@ def run_research( run_youtube: bool = False, run_tiktok: bool = False, run_instagram: bool = False, + run_xiaohongshu: bool = False, timeouts: dict = None, resolved_handle: str = None, do_hackernews: bool = True, @@ -737,6 +807,7 @@ def run_research( hackernews_error = None polymarket_error = None web_error = None + xiaohongshu_error = None # Determine web search mode do_web = sources in ("all", "web", "reddit-web", "x-web") @@ -764,6 +835,19 @@ def run_research( if progress: progress.start_web_only() progress.end_web_only() + # Optional Xiaohongshu search in web-only mode. + if run_xiaohongshu: + try: + xhs_items, xiaohongshu_error = _search_xiaohongshu( + topic, config, from_date, to_date, depth, + ) + web_items.extend(xhs_items) + if xiaohongshu_error and progress: + progress.show_error(f"Xiaohongshu error: {xiaohongshu_error}") + except Exception as e: + xiaohongshu_error = f"{type(e).__name__}: {e}" + if progress: + progress.show_error(f"Xiaohongshu error: {e}") # Still run YouTube/TikTok/Instagram in web-only mode if available if run_youtube: if progress: @@ -819,10 +903,20 @@ def run_research( youtube_future = None tiktok_future = None instagram_future = None + xiaohongshu_future = None hackernews_future = None polymarket_future = None web_future = None - max_workers = 2 + (1 if run_youtube else 0) + (1 if run_tiktok else 0) + (1 if run_instagram else 0) + (1 if do_hackernews else 0) + (1 if do_polymarket else 0) + (1 if web_backend else 0) + max_workers = ( + 2 + + (1 if run_youtube else 0) + + (1 if run_tiktok else 0) + + (1 if run_instagram else 0) + + (1 if run_xiaohongshu else 0) + + (1 if do_hackernews else 0) + + (1 if do_polymarket else 0) + + (1 if web_backend else 0) + ) with ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit searches @@ -865,6 +959,11 @@ def run_research( env.get_instagram_token(config), ) + if run_xiaohongshu: + xiaohongshu_future = executor.submit( + _search_xiaohongshu, topic, config, from_date, to_date, depth, + ) + if do_hackernews: if progress: progress.start_hackernews() @@ -971,6 +1070,21 @@ def run_research( if progress: progress.end_instagram(len(instagram_items)) + if xiaohongshu_future: + try: + xhs_items, xiaohongshu_error = xiaohongshu_future.result(timeout=future_timeout) + web_items.extend(xhs_items) + if xiaohongshu_error and progress: + progress.show_error(f"Xiaohongshu error: {xiaohongshu_error}") + except TimeoutError: + xiaohongshu_error = f"Xiaohongshu search timed out after {future_timeout}s" + if progress: + progress.show_error(xiaohongshu_error) + except Exception as e: + xiaohongshu_error = f"{type(e).__name__}: {e}" + if progress: + progress.show_error(f"Xiaohongshu error: {e}") + if hackernews_future: hn_timeout = timeouts.get("hackernews_future", future_timeout) try: @@ -1254,11 +1368,15 @@ def main(): # Auto-detect ScrapeCreators for Instagram has_instagram = env.is_instagram_available(config) + # Auto-detect Xiaohongshu HTTP API (requires service + login) + has_xiaohongshu = env.is_xiaohongshu_available(config) + # --diagnose: show source availability and exit if args.diagnose: web_source = env.get_web_search_source(config) diag = { "openai": bool(config.get("OPENAI_API_KEY")), + "reddit_public": True, "xai": bool(config.get("XAI_API_KEY")), "x_source": x_source_status["source"], "bird_installed": x_source_status["bird_installed"], @@ -1267,6 +1385,8 @@ def main(): "youtube": has_ytdlp, "tiktok": has_tiktok, "instagram": has_instagram, + "xiaohongshu": has_xiaohongshu, + "xiaohongshu_api_base": env.get_xiaohongshu_api_base(config), "hackernews": True, "polymarket": True, "web_search_backend": web_source, @@ -1290,6 +1410,7 @@ def main(): web_source = env.get_web_search_source(config) diag = { "openai": bool(config.get("OPENAI_API_KEY")), + "reddit_public": True, "xai": bool(config.get("XAI_API_KEY")), "x_source": x_source_status["source"], "bird_installed": x_source_status["bird_installed"], @@ -1297,6 +1418,8 @@ def main(): "bird_username": x_source_status.get("bird_username"), "youtube": has_ytdlp, "tiktok": has_tiktok, + "instagram": has_instagram, + "xiaohongshu": has_xiaohongshu, "hackernews": True, "polymarket": True, "web_search_backend": "deferred to assistant" if args.no_native_web else web_source, @@ -1383,6 +1506,7 @@ def main(): search_run_youtube = has_ytdlp search_run_tiktok = has_tiktok search_run_instagram = has_instagram + search_run_xiaohongshu = has_xiaohongshu if args.search: search_sources = parse_search_flag(args.search) has_reddit = "reddit" in search_sources @@ -1392,6 +1516,8 @@ def main(): search_run_youtube = "youtube" in search_sources and has_ytdlp search_run_tiktok = "tiktok" in search_sources and has_tiktok search_run_instagram = "instagram" in search_sources and has_instagram + # If explicitly requested, attempt Xiaohongshu even when preflight says unavailable. + search_run_xiaohongshu = "xiaohongshu" in search_sources include_search_web = "web" in search_sources # Map to existing sources string if has_reddit and has_x: @@ -1419,6 +1545,7 @@ def main(): run_youtube=search_run_youtube, run_tiktok=search_run_tiktok, run_instagram=search_run_instagram, + run_xiaohongshu=search_run_xiaohongshu, timeouts=timeouts, resolved_handle=args.x_handle, do_hackernews=search_do_hackernews, @@ -1541,8 +1668,6 @@ def main(): # Build source info for status footer source_info = {} - if not bool(config.get("OPENAI_API_KEY")): - source_info["reddit_skip_reason"] = "No OPENAI_API_KEY (add to ~/.config/last30days/.env)" if not x_source: if x_source_status["bird_installed"]: source_info["x_skip_reason"] = "Bird installed but not authenticated — log into x.com in browser" @@ -1556,6 +1681,11 @@ def main(): source_info["tiktok_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free credits)" if not has_instagram: source_info["instagram_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free credits)" + if not has_xiaohongshu: + source_info["xiaohongshu_skip_reason"] = ( + f"Xiaohongshu API unavailable or not logged in - start xiaohongshu-mcp and login " + f"(base: {env.get_xiaohongshu_api_base(config)})" + ) if not web_source: source_info["web_skip_reason"] = "assistant will use WebSearch (add BRAVE_API_KEY for native search)" diff --git a/scripts/lib/env.py b/scripts/lib/env.py index fcf7d2a..1d4febd 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -199,6 +199,7 @@ def get_config() -> Dict[str, Any]: ('OPENROUTER_API_KEY', None), ('PARALLEL_API_KEY', None), ('BRAVE_API_KEY', None), + ('XIAOHONGSHU_API_BASE', None), ('OPENAI_MODEL_POLICY', 'auto'), ('OPENAI_MODEL_PIN', None), ('XAI_MODEL_POLICY', 'latest'), @@ -221,24 +222,20 @@ def config_exists() -> bool: def get_available_sources(config: Dict[str, Any]) -> str: - """Determine which sources are available based on API keys. + """Determine which sources are available. Returns: 'all', 'both', 'reddit', 'reddit-web', 'x', 'x-web', 'web', or 'none' """ - has_openai = bool(config.get('OPENAI_API_KEY')) and config.get('OPENAI_AUTH_STATUS') == AUTH_STATUS_OK + # Reddit is available via public JSON fallback even without OpenAI auth. + has_reddit = True has_xai = bool(config.get('XAI_API_KEY')) has_web = has_web_search_keys(config) - if has_openai and has_xai: + if has_reddit and has_xai: return 'all' if has_web else 'both' - elif has_openai: + elif has_reddit: return 'reddit-web' if has_web else 'reddit' - elif has_xai: - return 'x-web' if has_web else 'x' - elif has_web: - return 'web' - else: - return 'web' # Fallback: assistant WebSearch (no API keys needed) + return 'web' if has_web else 'none' def has_web_search_keys(config: Dict[str, Any]) -> bool: @@ -267,7 +264,7 @@ def get_missing_keys(config: Dict[str, Any]) -> str: Returns: 'all', 'both', 'reddit', 'x', 'web', or 'none' """ - has_openai = bool(config.get('OPENAI_API_KEY')) and config.get('OPENAI_AUTH_STATUS') == AUTH_STATUS_OK + has_reddit = True has_xai = bool(config.get('XAI_API_KEY')) has_web = has_web_search_keys(config) @@ -277,16 +274,15 @@ def get_missing_keys(config: Dict[str, Any]) -> str: has_x = has_xai or has_bird - if has_openai and has_x and has_web: + if has_reddit and has_x and has_web: return 'none' - elif has_openai and has_x: + elif has_reddit and has_x: return 'web' # Missing web search keys - elif has_openai: + elif has_reddit and has_web: + return 'x' # Missing X source + elif has_reddit: return 'x' # Missing X source (and possibly web) - elif has_x: - return 'reddit' # Missing OpenAI key (and possibly web) - else: - return 'all' # Missing everything + return 'all' def validate_sources(requested: str, available: str, include_web: bool = False) -> tuple[str, Optional[str]]: @@ -300,56 +296,51 @@ def validate_sources(requested: str, available: str, include_web: bool = False) Returns: Tuple of (effective_sources, error_message) """ - # No API keys at all - if available == 'none': - if requested == 'auto': - return 'web', "No API keys configured. The assistant can still search the web if it has a search tool." - elif requested == 'web': - return 'web', None - else: - return 'web', f"No API keys configured. Add keys to ~/.config/last30days/.env for Reddit/X." - - # Web-only mode (only web search API keys) - if available == 'web': - if requested == 'auto': - return 'web', None - elif requested == 'web': - return 'web', None - else: - return 'web', "Only web search keys configured. Add OPENAI_API_KEY (or run codex login) for Reddit, XAI_API_KEY for X." + has_reddit = available in ('reddit', 'both', 'reddit-web', 'all') + has_x = available in ('x', 'both', 'x-web', 'all') + has_web = available in ('web', 'reddit-web', 'x-web', 'all') if requested == 'auto': - # Add web to sources if include_web is set + if has_reddit and has_x: + base = 'both' + elif has_reddit: + base = 'reddit' + elif has_x: + base = 'x' + elif has_web: + base = 'web' + else: + return 'none', "No sources are available." + if include_web: - if available == 'both': - return 'all', None # reddit + x + web - elif available == 'reddit': + if base == 'both': + return 'all', None + if base == 'reddit': return 'reddit-web', None - elif available == 'x': + if base == 'x': return 'x-web', None - return available, None + return base, None if requested == 'web': return 'web', None if requested == 'both': - if available not in ('both',): - missing = 'xAI' if available == 'reddit' else 'OpenAI' - return 'none', f"Requested both sources but {missing} key is missing. Use --sources=auto to use available keys." + if not (has_reddit and has_x): + return 'none', "Requested both sources but X source is missing." if include_web: return 'all', None return 'both', None if requested == 'reddit': - if available == 'x': + if not has_reddit: return 'none', "Requested Reddit but only xAI key is available." if include_web: return 'reddit-web', None return 'reddit', None if requested == 'x': - if available == 'reddit': - return 'none', "Requested X but only OpenAI key is available." + if not has_x: + return 'none', "Requested X but no X source is available (need Bird auth or XAI_API_KEY)." if include_web: return 'x-web', None return 'x', None @@ -435,6 +426,40 @@ def get_instagram_token(config: Dict[str, Any]) -> str: return config.get('SCRAPECREATORS_API_KEY') or '' +def get_xiaohongshu_api_base(config: Dict[str, Any]) -> str: + """Get Xiaohongshu HTTP API base URL. + + Defaults to host.docker.internal so OpenClaw Docker can reach host service. + """ + return (config.get('XIAOHONGSHU_API_BASE') or "http://host.docker.internal:18060").rstrip("/") + + +def is_xiaohongshu_available(config: Dict[str, Any]) -> bool: + """Check whether Xiaohongshu HTTP API is reachable and logged in.""" + # Import here to avoid heavy imports at module load. + from . import http + + base = get_xiaohongshu_api_base(config) + try: + # Keep health probe snappy, but allow one retry for transient hiccups. + health = http.get(f"{base}/health", timeout=3, retries=2) + if not isinstance(health, dict): + return False + if not health.get("success"): + return False + + # Login probe can be slower on some deployments (browser/session checks), + # so use a slightly longer timeout to avoid false negatives. + login = http.get(f"{base}/api/v1/login/status", timeout=8, retries=2) + is_logged_in = ( + login.get("data", {}).get("is_logged_in") + if isinstance(login, dict) else False + ) + return bool(is_logged_in) + except Exception: + return False + + # Backward compat alias is_apify_available = is_tiktok_available diff --git a/scripts/lib/openai_reddit.py b/scripts/lib/openai_reddit.py index 6a8f361..f98e8e4 100644 --- a/scripts/lib/openai_reddit.py +++ b/scripts/lib/openai_reddit.py @@ -355,6 +355,105 @@ def search_reddit( raise http.HTTPError("No models available") +def _public_relevance(score: int, num_comments: int) -> float: + """Estimate relevance for public Reddit search results.""" + # Lightweight heuristic: blend normalized score + comments. + score_component = min(1.0, max(0.0, score / 500.0)) + comments_component = min(1.0, max(0.0, num_comments / 200.0)) + return round((score_component * 0.6) + (comments_component * 0.4), 3) + + +def search_reddit_public( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", +) -> List[Dict[str, Any]]: + """Search Reddit directly via public JSON endpoint (no OpenAI key required). + + This is a fallback mode for environments where OpenAI auth is unavailable. + It uses reddit.com/search/.json with recency filter (t=month). + """ + _, max_items = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + limit = min(100, max(20, max_items)) + + core = _extract_core_subject(topic) + queries = [topic] + if core and core.lower() != topic.lower(): + queries.append(core) + queries.append(f'"{core}"') + + seen_urls = set() + all_items: List[Dict[str, Any]] = [] + + headers = { + "User-Agent": http.USER_AGENT, + "Accept": "application/json", + } + + for query in queries: + try: + url = ( + "https://www.reddit.com/search/.json" + f"?q={_url_encode(query)}&sort=new&t=month&limit={limit}&raw_json=1" + ) + data = http.get(url, headers=headers, timeout=20, retries=2) + children = data.get("data", {}).get("children", []) + for child in children: + if child.get("kind") != "t3": + continue + post = child.get("data", {}) + permalink = str(post.get("permalink", "")).strip() + if not permalink or "/comments/" not in permalink: + continue + + full_url = f"https://www.reddit.com{permalink}" + if full_url in seen_urls: + continue + seen_urls.add(full_url) + + score = int(post.get("score", 0) or 0) + num_comments = int(post.get("num_comments", 0) or 0) + + # Parse date from created_utc + created_utc = post.get("created_utc") + date_value = None + if created_utc: + from . import dates as dates_mod + date_value = dates_mod.timestamp_to_date(created_utc) + + all_items.append({ + "id": f"R{len(all_items)+1}", + "title": str(post.get("title", "")).strip(), + "url": full_url, + "subreddit": str(post.get("subreddit", "")).strip(), + "date": date_value, + "why_relevant": "Found via Reddit public search", + "relevance": _public_relevance(score, num_comments), + "engagement": { + "score": score, + "num_comments": num_comments, + "upvote_ratio": post.get("upvote_ratio"), + }, + }) + + except http.HTTPError as e: + _log_info(f"Public Reddit search failed for query '{query}': {e}") + # Continue with next query; partial results are still useful. + continue + except Exception as e: + _log_info(f"Public Reddit search error for query '{query}': {e}") + continue + + # Sort by date (desc, unknown dates last), then relevance desc + def _sort_key(item: Dict[str, Any]): + date_str = item.get("date") or "" + return (date_str, float(item.get("relevance", 0.0))) + + all_items.sort(key=_sort_key, reverse=True) + return all_items[: max_items * 2] + + def search_subreddits( subreddits: List[str], topic: str, diff --git a/scripts/lib/render.py b/scripts/lib/render.py index 510135a..848f3f6 100644 --- a/scripts/lib/render.py +++ b/scripts/lib/render.py @@ -502,6 +502,20 @@ def render_source_status(report: schema.Report, source_info: dict = None) -> str lines.append(f" ✅ Instagram: {len(report.instagram)} reels ({with_captions} with captions)") # Hide when zero results + # Xiaohongshu (from Web source bucket) + xhs_count = 0 + if report.web: + xhs_count = sum( + 1 for w in report.web + if getattr(w, "source_domain", "").lower().endswith("xiaohongshu.com") + ) + if xhs_count > 0: + lines.append(f" ✅ Xiaohongshu: {xhs_count} notes") + else: + reason = source_info.get("xiaohongshu_skip_reason") + if reason: + lines.append(f" ⚡ Xiaohongshu: {reason}") + # Hacker News if report.hackernews_error: lines.append(f" ❌ HN: error - {report.hackernews_error}") diff --git a/scripts/lib/ui.py b/scripts/lib/ui.py index 1aeff0b..2351344 100644 --- a/scripts/lib/ui.py +++ b/scripts/lib/ui.py @@ -426,12 +426,15 @@ def show_diagnostic_banner(diag: dict): bird_username, youtube, web_search_backend """ has_openai = diag.get("openai", False) + has_reddit_public = diag.get("reddit_public", False) + has_reddit = has_openai or has_reddit_public has_x = diag.get("x_source") is not None has_youtube = diag.get("youtube", False) + has_xiaohongshu = diag.get("xiaohongshu", False) has_web = diag.get("web_search_backend") is not None # If everything is available, no banner needed - if has_openai and has_x and has_youtube and has_web: + if has_reddit and has_x and has_youtube and has_web: return lines = [] @@ -443,7 +446,9 @@ def show_diagnostic_banner(diag: dict): # Reddit if has_openai: - lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Reddit{Colors.RESET} — OPENAI_API_KEY found {Colors.DIM}│{Colors.RESET}") + lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Reddit{Colors.RESET} — OpenAI/Codex auth found {Colors.DIM}│{Colors.RESET}") + elif has_reddit_public: + lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Reddit{Colors.RESET} — Public Reddit search (no key) {Colors.DIM}│{Colors.RESET}") else: lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.RED}❌ Reddit{Colors.RESET} — No OPENAI_API_KEY {Colors.DIM}│{Colors.RESET}") lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Add to ~/.config/last30days/.env {Colors.DIM}│{Colors.RESET}") @@ -469,6 +474,12 @@ def show_diagnostic_banner(diag: dict): lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.RED}❌ YouTube{Colors.RESET} — yt-dlp not installed {Colors.DIM}│{Colors.RESET}") lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Fix: brew install yt-dlp (free) {Colors.DIM}│{Colors.RESET}") + # Xiaohongshu + if has_xiaohongshu: + lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Xiaohongshu{Colors.RESET} — API connected + logged in {Colors.DIM}│{Colors.RESET}") + else: + lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.YELLOW}⚡ Xiaohongshu{Colors.RESET} — API not connected/logged in {Colors.DIM}│{Colors.RESET}") + # Web if has_web: backend = diag.get("web_search_backend", "") @@ -486,7 +497,9 @@ def show_diagnostic_banner(diag: dict): lines.append("│ │") if has_openai: - lines.append("│ ✅ Reddit — OPENAI_API_KEY found │") + lines.append("│ ✅ Reddit — OpenAI/Codex auth found │") + elif has_reddit_public: + lines.append("│ ✅ Reddit — Public Reddit search (no key) │") else: lines.append("│ ❌ Reddit — No OPENAI_API_KEY │") lines.append("│ └─ Add to ~/.config/last30days/.env │") @@ -506,6 +519,11 @@ def show_diagnostic_banner(diag: dict): lines.append("│ ❌ YouTube — yt-dlp not installed │") lines.append("│ └─ Fix: brew install yt-dlp (free) │") + if has_xiaohongshu: + lines.append("│ ✅ Xiaohongshu — API connected + logged in │") + else: + lines.append("│ ⚡ Xiaohongshu — API not connected/logged in │") + if has_web: lines.append("│ ✅ Web — API search available │") else: diff --git a/scripts/lib/xiaohongshu_api.py b/scripts/lib/xiaohongshu_api.py new file mode 100644 index 0000000..12cee07 --- /dev/null +++ b/scripts/lib/xiaohongshu_api.py @@ -0,0 +1,162 @@ +"""Xiaohongshu HTTP API search client for last30days. + +Uses xpzouying/xiaohongshu-mcp REST endpoints: +- GET/POST /api/v1/feeds/search +- GET /api/v1/login/status +""" + +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from . import http + + +def _to_int(value: Any) -> int: + """Convert Xiaohongshu count strings to int. + + Supports plain ints and Chinese suffixes like 1.2万 / 3亿. + """ + if value is None: + return 0 + if isinstance(value, (int, float)): + return int(value) + + text = str(value).strip().lower().replace(",", "") + if not text: + return 0 + + try: + if text.endswith("万"): + return int(float(text[:-1]) * 10000) + if text.endswith("亿"): + return int(float(text[:-1]) * 100000000) + return int(float(text)) + except (TypeError, ValueError): + return 0 + + +def _timestamp_to_date_ms(ts: Any) -> Optional[str]: + """Convert millisecond timestamp to YYYY-MM-DD.""" + try: + iv = int(ts) + if iv <= 0: + return None + # API examples use milliseconds. + dt = datetime.fromtimestamp(iv / 1000.0, tz=timezone.utc) + return dt.strftime("%Y-%m-%d") + except (TypeError, ValueError, OSError): + return None + + +def _relevance_from_interactions(likes: int, comments: int, favorites: int) -> float: + """Heuristic relevance score from engagement metrics.""" + # Weighted engagement with soft caps to [0, 1]. + weighted = (likes * 1.0) + (comments * 2.5) + (favorites * 1.5) + # 5000 weighted engagement ~= strong relevance. + score = min(1.0, max(0.05, weighted / 5000.0)) + return round(score, 3) + + +def _build_note_url(feed_id: str, xsec_token: str) -> str: + """Build a stable Xiaohongshu note URL.""" + if xsec_token: + return f"https://www.xiaohongshu.com/explore/{feed_id}?xsec_token={xsec_token}" + return f"https://www.xiaohongshu.com/explore/{feed_id}" + + +def search_feeds( + topic: str, + from_date: str, + to_date: str, + base_url: str, + depth: str = "default", +) -> List[Dict[str, Any]]: + """Search Xiaohongshu feeds and normalize to web-item shape.""" + base = (base_url or "").rstrip("/") + if not base: + raise ValueError("Missing Xiaohongshu API base URL") + + # Quick login sanity check. + login = http.get(f"{base}/api/v1/login/status", timeout=8, retries=1) + is_logged_in = ( + login.get("data", {}).get("is_logged_in") + if isinstance(login, dict) else False + ) + if not is_logged_in: + raise http.HTTPError("Xiaohongshu API reachable but not logged in") + + # API supports filters; use recency-oriented defaults. + publish_time = "一天内" if depth == "quick" else "一周内" if depth == "default" else "半年内" + payload = { + "keyword": topic, + "filters": { + "sort_by": "综合", + "note_type": "不限", + "publish_time": publish_time, + "search_scope": "不限", + "location": "不限", + }, + } + + resp = http.post(f"{base}/api/v1/feeds/search", payload, timeout=20, retries=1) + feeds = resp.get("data", {}).get("feeds", []) if isinstance(resp, dict) else [] + if not isinstance(feeds, list): + feeds = [] + + # Cap source volume similarly to other web sources. + limit = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15) + items: List[Dict[str, Any]] = [] + + for i, feed in enumerate(feeds[:limit]): + if not isinstance(feed, dict): + continue + note = feed.get("noteCard") or {} + if not isinstance(note, dict): + note = {} + interact = note.get("interactInfo") or {} + if not isinstance(interact, dict): + interact = {} + + feed_id = str(feed.get("id") or note.get("noteId") or "").strip() + if not feed_id: + continue + + xsec_token = str(feed.get("xsecToken") or note.get("xsecToken") or "").strip() + title = str( + note.get("displayTitle") + or note.get("title") + or "" + ).strip() + snippet = str( + note.get("desc") + or note.get("displayDesc") + or title + or "" + ).strip() + + likes = _to_int(interact.get("likedCount")) + comments = _to_int(interact.get("commentCount")) + favorites = _to_int(interact.get("collectedCount")) + + date_value = _timestamp_to_date_ms(note.get("time")) + why = f"Xiaohongshu engagement: likes={likes}, comments={comments}, favorites={favorites}" + + items.append({ + "id": f"XHS{i+1}", + "title": title[:200] if title else f"Xiaohongshu note {feed_id}", + "url": _build_note_url(feed_id, xsec_token), + "source_domain": "xiaohongshu.com", + "snippet": snippet[:500], + "date": date_value, + "date_confidence": "high" if date_value else "low", + "relevance": _relevance_from_interactions(likes, comments, favorites), + "why_relevant": why, + # Keep raw engagement for debugging/possible future rendering. + "engagement": { + "likes": likes, + "comments": comments, + "favorites": favorites, + }, + }) + + return items