Upload data1/main_v2.py with huggingface_hub

data1/main_v2.py

@@ -0,0 +1,1007 @@
+# main.py
+import os
+import time
+
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
+
+from typing import List, Dict, Any, Optional
+from pydantic import BaseModel
+import pandas as pd
+import shutil
+import subprocess
+import requests
+from requests.exceptions import RequestException, Timeout
+import argparse
+import asyncio
+from pathlib import Path
+from dotenv import load_dotenv
+from datetime import datetime, timedelta
+from util import init_logger, logger, call_llm, CODE_EXTENSIONS, extract_final_answer_from_reasoning
+from langchain_core.output_parsers import JsonOutputParser
+
+
+# Pydantic models for structured output
+class RelevanceResult(BaseModel):
+    relevant: str  # YES or NO
+    reason: str
+
+
+class ExpandedKeywords(BaseModel):
+    keywords: List[str]
+
+
+# GitHub 请求的超时与重试配置（可通过环境变量覆盖）
+GITHUB_TIMEOUT = int(os.environ.get("GITHUB_TIMEOUT", "30"))  # 单次请求超时秒数
+GITHUB_MAX_RETRIES = int(os.environ.get("GITHUB_MAX_RETRIES", "3"))  # 最多重试次数
+GITHUB_RETRY_BACKOFF = float(os.environ.get("GITHUB_RETRY_BACKOFF", "1.5"))  # 指数回退基数
+
+
+def github_get(
+    url: str,
+    *,
+    headers: Optional[Dict[str, str]] = None,
+    params: Optional[Dict[str, Any]] = None,
+    timeout: Optional[int] = None,
+) -> Optional[requests.Response]:
+    """
+    带超时和重试的 GitHub GET 请求封装。
+    - 使用环境变量控制超时和重试次数
+    - 对 Timeout / RequestException 进行重试
+    """
+    if timeout is None:
+        timeout = GITHUB_TIMEOUT
+
+    last_exc: Optional[BaseException] = None
+
+    for attempt in range(1, GITHUB_MAX_RETRIES + 1):
+        try:
+            resp = requests.get(url, headers=headers, params=params, timeout=timeout)
+            return resp
+        except (Timeout, RequestException) as e:
+            last_exc = e
+            logger.warning(
+                f"GitHub 请求失败（第 {attempt}/{GITHUB_MAX_RETRIES} 次）: {url} | {e}"
+            )
+            if attempt < GITHUB_MAX_RETRIES:
+                # 指数回退：1, 1.5, 2.25, ...
+                sleep_s = GITHUB_RETRY_BACKOFF ** (attempt - 1)
+                time.sleep(sleep_s)
+
+    logger.error(f"GitHub 请求多次失败，放弃: {url} | 最后错误: {last_exc}")
+    return None
+
+
+def search_github_repos(keywords: List[str], token: str, output_csv: Path):
+    """Search GitHub repos with incremental write per keyword using pending logic"""
+    headers = {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json"}
+
+    # Calculate pending keywords using set difference
+    all_keywords = set(k.lower() for k in keywords)
+    searched_keywords = set()
+
+    if output_csv.exists():
+        df_existing = pd.read_csv(output_csv)
+        searched_keywords = set(df_existing["keyword"].str.lower().unique())
+        logger.info(f"Resume: Already searched {len(searched_keywords)} keywords")
+        return
+
+    # Pending keywords = all - searched
+    pending_keywords = all_keywords - searched_keywords
+    pending = [k for k in keywords if k.lower() in pending_keywords]
+
+    logger.info(f"Pending: {len(pending)} keywords to search")
+
+    # Load existing URLs for global deduplication
+    global_seen = set()
+    if output_csv.exists():
+        global_seen = set(df_existing["url"].tolist())
+
+    def get_count(query: str) -> int:
+        """Get total count without fetching data (with timeout & retries)"""
+        try:
+            resp = github_get(
+                "https://api.github.com/search/repositories",
+                headers=headers,
+                params={"q": query, "per_page": 1},
+            )
+            if resp is not None and resp.status_code == 200:
+                return resp.json().get("total_count", 0)
+        except Exception as e:
+            logger.error(f"get_count error for query '{query}': {e}")
+        return 0
+
+    def fetch_repos(query: str, keyword: str, local_repos: List[Dict]):
+        """Fetch all results for a single query (with timeout & retries)"""
+        page = 1
+        while page <= 10:
+            try:
+                resp = github_get(
+                    "https://api.github.com/search/repositories",
+                    headers=headers,
+                    params={"q": query, "per_page": 100, "page": page},
+                )
+                if resp is None:
+                    # 多次重试失败，放弃当前 query
+                    logger.error(f"Fetch error: all retries failed for query '{query}', page {page}")
+                    break
+                if resp.status_code != 200:
+                    logger.error(
+                        f"Fetch error: status_code={resp.status_code} for query '{query}', page {page}"
+                    )
+                    break
+                items = resp.json().get("items", [])
+                if not items:
+                    break
+
+                for r in items:
+                    url = r.get("html_url", "")
+                    if url and url not in global_seen:
+                        global_seen.add(url)
+                        repo_data = {
+                            "keyword": keyword,
+                            "name": r.get("name", ""),
+                            "full_name": r.get("full_name", ""),
+                            "owner": r.get("owner", {}).get("login", ""),
+                            "url": url,
+                            "description": r.get("description") or "",
+                            "language": r.get("language") or "",
+                            "topics": ",".join(r.get("topics", [])),
+                            "stars": r.get("stargazers_count", 0),
+                            "forks": r.get("forks_count", 0),
+                            "created_at": r.get("created_at", ""),
+                            "updated_at": r.get("updated_at", ""),
+                            "pushed_at": r.get("pushed_at", ""),
+                            "license": r.get("license", {}).get("spdx_id", "") if r.get("license") else "",
+                            "default_branch": r.get("default_branch", ""),
+                            "open_issues": r.get("open_issues_count", 0),
+                            "size": r.get("size", 0),
+                            "has_wiki": r.get("has_wiki", False),
+                            "archived": r.get("archived", False),
+                        }
+                        local_repos.append(repo_data)
+
+                if len(items) < 100:
+                    break
+                page += 1
+            except Exception as e:
+                logger.error(f"Fetch error (unexpected): {e}")
+                break
+
+    def split_by_date(kw: str, keyword: str, start_date: datetime, end_date: datetime, local_repos: List[Dict]):
+        """Recursive date splitting with stars>10 and in:readme filters"""
+        start_str = start_date.strftime("%Y-%m-%d")
+        end_str = end_date.strftime("%Y-%m-%d")
+        query = f"{kw} in:readme stars:>10 created:{start_str}..{end_str}"
+        count = get_count(query)
+        logger.info(f"  {start_str} to {end_str}: {count} repos")
+
+        if count == 0:
+            return
+        elif count <= 1000:
+            fetch_repos(query, keyword, local_repos)
+        else:
+            days = (end_date - start_date).days
+            if days == 0:
+                logger.warning(f"Single day has {count} repos, getting first 1000: {start_str}")
+                fetch_repos(query, keyword, local_repos)
+            else:
+                mid_days = days // 2
+                mid_date = start_date + timedelta(days=mid_days)
+                split_by_date(kw, keyword, start_date, mid_date, local_repos)
+                split_by_date(kw, keyword, mid_date + timedelta(days=1), end_date, local_repos)
+
+    # Search each pending keyword and write immediately
+    for kw in pending:
+        logger.info(f"Searching keyword: {kw}")
+        keyword_repos = []
+        start = datetime(2008, 1, 1)
+        end = datetime.now()
+        split_by_date(kw, kw, start, end, keyword_repos)
+
+        # Write immediately after each keyword
+        if keyword_repos:
+            df_new = pd.DataFrame(keyword_repos)
+            df_new.to_csv(output_csv, mode="a", header=not output_csv.exists(), index=False, encoding="utf-8")
+            logger.info(f"✓ Saved {len(keyword_repos)} repos for keyword: {kw}")
+        else:
+            logger.info(f"✓ No new repos for keyword: {kw}")
+
+    logger.info(f"Total repos in CSV: {len(global_seen)}")
+
+
+async def get_readme(owner: str, repo: str, token: str) -> str:
+    """Fetch README content from repo (async with timeout & retries)"""
+    try:
+        # 使用 asyncio.to_thread 将同步的 github_get 包装为异步，避免阻塞事件循环
+        resp = await asyncio.to_thread(
+            github_get,
+            f"https://api.github.com/repos/{owner}/{repo}/readme",
+            headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3.raw"},
+        )
+        if resp is not None and resp.status_code == 200:
+            return resp.text
+        return ""
+    except Exception as e:
+        logger.error(f"get_readme error for {owner}/{repo}: {e}")
+        return ""
+
+
+async def check_relevance(
+    repo: Dict, keywords: List[str], model: str, base_url: str, api_key: str, token: str, log_file: str
+) -> bool:
+    """Use LLM to check if repo is relevant to keywords"""
+    readme = get_readme(repo["owner"], repo["name"], token)[:8000]
+
+    prompt = f"""Determine if this GitHub repository is relevant to the keywords: {', '.join(keywords)}
+
+Repository: {repo['name']}
+Description: {repo['description']}
+Language: {repo['language']}
+README (truncated):
+{readme}
+
+Answer 'YES' if the repository is related to any of the keywords, 'NO' otherwise.
+Provide your reasoning in the reason field."""
+
+    try:
+        result = await call_llm(
+            [{"role": "user", "content": prompt}],
+            model,
+            base_url,
+            api_key,
+            pydantic_object=RelevanceResult,
+            log_file=log_file,
+            temperature=0.1,
+        )
+        return result.get("relevant", "").upper() == "YES"
+    except Exception as e:
+        logger.error(f"LLM error for {repo['name']}: {e}")
+        return False
+
+
+def save_csv(repos: List[Dict], path: str):
+    """Save repos to CSV using pandas"""
+    df = pd.DataFrame(repos)
+    df.to_csv(path, index=False, encoding="utf-8")
+    logger.info(f"Saved {len(repos)} repos to {path}")
+
+
+def clone_repos_batch(repos: List[Dict], dest_dir: Path) -> List[str]:
+    """Clone a batch of repos, return list of successfully cloned full_names"""
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    cloned = []
+    
+    for row in repos:
+        full_name = row["full_name"]
+        repo_path = dest_dir / full_name.replace("/", "___")  # three underscores
+        try:
+            subprocess.run(
+                ["git", "clone", "--depth", "1", row["url"], str(repo_path)], 
+                check=True, 
+                capture_output=True,
+                timeout=600  # 10 minutes timeout
+            )
+            cloned.append(full_name)
+            logger.info(f"✓ Cloned: {full_name}")
+        except subprocess.TimeoutExpired:
+            logger.error(f"✗ Clone timeout: {full_name}")
+        except Exception as e:
+            logger.error(f"✗ Clone failed {full_name}: {e}")
+    
+    return cloned
+
+
+def filter_code_files(repo_dir: Path, dest_repo: Path) -> int:
+    """Filter code files from a repo directory, return file count"""
+    # NOTE:
+    # - 过滤目录用于后续处理/打开查看；超大文件（尤其是 .ipynb）会导致 IDE/工具无法打开（常见上限 50MB）
+    # - 因此这里支持最大单文件大小限制，避免复制超大文件
+    file_count = 0
+    
+    for root, dirs, files in os.walk(repo_dir):
+        # Skip hidden and common build/dependency directories
+        dirs[:] = [
+            d
+            for d in dirs
+            if not d.startswith(".")
+            and d not in {"node_modules", "__pycache__", "venv", ".git", "build", "dist", "target"}
+        ]
+        
+        for f in files:
+            src = Path(root) / f
+            if src.suffix.lower() in CODE_EXTENSIONS and src.exists():
+                # 默认不限制文件大小（保留超大文件），避免误删重要数据。
+                # 如需限制，可设置环境变量 MAX_FILTER_FILE_SIZE_BYTES（单位 bytes）。
+                # 约定：<=0 表示不限制。
+                try:
+                    max_bytes = int(os.environ.get("MAX_FILTER_FILE_SIZE_BYTES", "0"))
+                except Exception:
+                    max_bytes = 0
+
+                try:
+                    if max_bytes > 0:
+                        size = src.stat().st_size
+                        if size > max_bytes:
+                            logger.info(
+                                f"Skip large file (> {max_bytes} bytes): {size} bytes | {src}"
+                            )
+                            continue
+                except Exception as e:
+                    logger.warning(f"Failed to stat {src}, skip: {e}")
+                    continue
+
+                rel = src.relative_to(repo_dir)
+                dst = dest_repo / rel
+                dst.parent.mkdir(parents=True, exist_ok=True)
+                try:
+                    shutil.copy2(src, dst)
+                    file_count += 1
+                except Exception as e:
+                    logger.warning(f"Failed to copy {src}: {e}")
+    
+    return file_count
+
+
+def process_repos_batch(
+    repos: List[Dict], 
+    batch_dir: Path, 
+    filtered_dir: Path,
+    processed_csv: Path
+) -> List[str]:
+    """
+    Process a batch of repos: clone -> filter -> delete
+    Returns list of successfully processed full_names
+    """
+    # Clone batch
+    cloned_fullnames = clone_repos_batch(repos, batch_dir)
+    
+    if not cloned_fullnames:
+        return []
+    
+    # Filter code files for each cloned repo
+    processed = []
+    for full_name in cloned_fullnames:
+        repo_name = full_name.replace("/", "___")
+        repo_path = batch_dir / repo_name
+        dest_repo = filtered_dir / repo_name
+        
+        if not repo_path.exists():
+            continue
+            
+        file_count = filter_code_files(repo_path, dest_repo)
+        
+        if file_count > 0:
+            processed.append(full_name)
+            logger.info(f"✓ Processed {full_name}: {file_count} code files")
+        else:
+            # No code files found, remove empty destination
+            if dest_repo.exists():
+                shutil.rmtree(dest_repo)
+            logger.info(f"✗ No code files in {full_name}")
+    
+    # Delete cloned repos to save space
+    for full_name in cloned_fullnames:
+        repo_name = full_name.replace("/", "___")
+        repo_path = batch_dir / repo_name
+        if repo_path.exists():
+            try:
+                shutil.rmtree(repo_path)
+                logger.debug(f"Deleted {repo_name}")
+            except Exception as e:
+                logger.warning(f"Failed to delete {repo_name}: {e}")
+    
+    # Record processed repos
+    if processed:
+        # Create mapping from full_name to repo dict
+        repo_map = {r["full_name"]: r for r in repos}
+        processed_records = [
+            {"url": repo_map[fn]["url"], "full_name": fn} 
+            for fn in processed 
+            if fn in repo_map
+        ]
+        if processed_records:
+            df_processed = pd.DataFrame(processed_records)
+            df_processed.to_csv(processed_csv, mode="a", header=not processed_csv.exists(), index=False, encoding="utf-8")
+    
+    return processed
+
+
+async def main():
+    load_dotenv()
+    parser = argparse.ArgumentParser(description="GitHub Repo Crawler")
+    parser.add_argument(
+        "--mode",
+        type=str,
+        default="all",
+        choices=["all", "step2", "step34"],
+        help="运行模式：all=步骤2与步骤3&4并行；step2=仅相关性检查；step34=仅克隆+过滤",
+    )
+    parser.add_argument(
+        "--watch",
+        action="store_true",
+        help="仅对 --mode step34 有效：持续轮询 repos_check_history.csv 并处理新的 YES（不加则只跑一轮就退出）",
+    )
+    parser.add_argument(
+        "--poll_interval",
+        type=int,
+        default=30,
+        help="仅对 --mode step34 --watch 有效：轮询间隔（秒）",
+    )
+    parser.add_argument(
+        "--max_idle",
+        type=int,
+        default=20,
+        help="仅对 --mode step34 --watch 有效：连续空转次数上限（每次间隔 poll_interval 秒），超过后退出",
+    )
+    parser.add_argument(
+        "--keywords",
+        type=str,
+        default="Chemistry, Biology, Biochemistry, Omics, Medicine, Pharmacology, Toxicology, Bioinformatics, Bioengineering, Biophysics, Viral, Microbial, Prediction, Discovery, Protein, Gene, DNA, RNA, Vaccine, Computational Biology, Computational Biochemistry, Computational Chemistry, Computational Materials, Quantum Chemistry, Disease, Biomedical, Material, Pharmacogenetics, Pharmacogenomics, Modeling, Networks, In Silico, Pathology, Physiology, Genomics, Proteomics, Transcriptomics, Metabolomics, Glycomics, Lipidomics, Immunology, Microbiology, Molecular biology, Pharmaceutics, Network pharmacology, Epigenetics, Sequencing, Design, Multi-omics, Biomarker, System biology, Synthetic biology, Cell biology, Cancer biology, Ensemble, Personalized, Lipid, Metabolic, Genesis, Ion, Heterogeneity, Generative, Generate, Human, Receptor, Ligand, Organoid, Evolution, Pathogens, Homeostasis, Allele, Genotype, Phenotype, Antibody, Antigen, Nucleic acids, Carbohydrate, Substrate, Inhibition, Activation, Allosteric, Cofactor, Coenzyme, Enzyme, Redox, Hydrophilic, Hydrophobic, Codon, Transcription, Translation, Pathway, Cycle, Signaling, Dynamics, Kinetics, Docking, Spectrometry, Profiling, Diagnostics, CRISPR, Bio, Marker, Pharmacokinetics, Pharmacodynamics, Absorption, Mechanism of action, Agonist, Antagonist, Bioavailability, Half-life, Reaction, Drug, Biologics, Pharmacometrics, Beta-blocker, Regulatory networks, Multi-scale modeling, Single-cell, Spatial biology, Integration, Monte Carlo, System immunology, Metagenomics, QSAR, QAPR, Chemical space, AlphaFold, Folding, Mechanism, Digital twin, Virtual human, Gene editing, Bio foundation model, Biotechnology, Assay, Lead discovery, High-throughput, Screening, Hit-to-lead, Lead optimization, De novo, ADMET, Translational medicine, Drug repurpose, Conjugate, Agent-based model, Compartmental model, Reproduction number, Nowcasting, Phylodynamic model, Physiologically based pharmacokinetics model, PBPK model, Organ-on-a-chip, Anomaly detection, Stochastic modeling, Genomic surveillance, Antimicrobial resistance modeling, AMR, Pandemic, Digital PCR, Next-generation sequencing, Biosensors, Imaging, Sensors, Quantum mechanics, DFT, Ab initio, Hartree-Fock, Coupled cluster, Electronic structure, Homo-Lumo, Conformation, Cheminformatics, QM/MM, First-principles based DFT, Diffusion, Finite element method, Phase-field technique, Potential, Metamaterial, 2D, 3D, Porous, Crystal, Rosettafold, Gene regulatory networks, Cell atlas, Human atlas, Spatial transcriptomics, Pseudotime analysis, Quantum biology, Metabolic flux analysis, Free energy perturbation, Protein-protein, Explainable AI, Neurology, Reinforcement learning, Generative AI, Flow matching, Generative adversarial networks, GAN, Variational autoencoders, VAE, Autoregressive, Transformer, Recurrent neural networks, RNN, Score",
+        help="Comma-separated keywords",
+    )
+    parser.add_argument("--workdir", type=str, default="./workdir", help="Working directory")
+    parser.add_argument("--model", type=str, default=os.getenv("OPENAI_MODEL", "gpt-4o"))
+    parser.add_argument("--base_url", type=str, default=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), 
+                        help="Base URL(s) for VLLM service(s). For multiple GPUs, use comma-separated URLs (e.g., 'http://gpu1:8000/v1,http://gpu2:8000/v1')")
+    parser.add_argument("--api_key", type=str, default=os.getenv("OPENAI_API_KEY"))
+    parser.add_argument("--batch_size", type=int, default=10, help="Number of repos to process in each batch")
+    parser.add_argument("--max_tokens", type=int, default=1000, help="Maximum tokens for LLM generation in relevance check (Step 2)")
+    args = parser.parse_args()
+    # Unified config
+    workdir = Path(args.workdir)
+    # Parse base_urls (support multiple VLLM services for parallel processing)
+    base_urls = [url.strip() for url in args.base_url.split(",") if url.strip()]
+    if not base_urls:
+        base_urls = ["https://api.openai.com/v1"]  # Default fallback
+    
+    config = {
+        "workdir": workdir,
+        "keywords_expanded": workdir / "keywords_expanded.json",
+        "repos_searched": workdir / "repos_searched.csv",
+        "repos_checked": workdir / "repos_checked.csv",
+        "repos_raw": workdir / "repos_raw",  # Temporary batch directory
+        "repos_filtered": workdir / "repos_filtered",
+        "repos_processed": workdir / "repos_processed.csv",  # Track processed repos
+        "log_file": str(workdir / "calls_llm.jsonl"),
+        "model": args.model,
+        "base_url": args.base_url,  # Keep for backward compatibility
+        "base_urls": base_urls,  # List of VLLM service URLs
+        "api_key": args.api_key,
+        "github_token": os.environ.get("GITHUB_TOKEN"),
+        "keywords": [k.strip() for k in args.keywords.split(",") if k.strip()],
+        "batch_size": args.batch_size,
+        "max_tokens": args.max_tokens,
+    }
+
+    # Setup
+    os.makedirs(config["workdir"], exist_ok=True)
+    init_logger(str(config["workdir"] / "run.log"))
+    logger.info(f"Base keywords: {config['keywords']}")
+    logger.info(f"Model: {config['model']}")
+    logger.info(f"Run mode: {args.mode} | step34.watch={args.watch} | poll_interval={args.poll_interval} | max_idle={args.max_idle}")
+
+    # mode=step34 时，不需要跑 Step0/1/2；只要依赖 repos_check_history + repos_searched（用于补全字段）即可
+    expanded = None
+    if args.mode in {"all", "step2"}:
+        # Step 0: Expand keywords with LLM (skip if keywords_expanded.json exists)
+        if config["keywords_expanded"].exists():
+            logger.info(f"[Skip] Step 0: {config['keywords_expanded']} exists")
+            import json
+
+            with open(config["keywords_expanded"], "r") as f:
+                expanded = json.load(f)["keywords"]
+        else:
+            logger.info("=" * 60 + "\nStep 0: Expand Keywords with LLM\n" + "=" * 60)
+            parser0 = JsonOutputParser(pydantic_object=ExpandedKeywords)
+            messages = [
+                {
+                    "role": "system",
+                    "content": f"You are an assistant that generates diverse and related keywords for scientific disciplines.\n{parser0.get_format_instructions()}",
+                },
+                {
+                    "role": "user",
+                    "content": f"""Generate a list of exactly 5 diverse keywords related to these scientific fields: {', '.join(config['keywords'])}.
+Make sure that the generated keywords do not stray away from these scientific disciplines and do not contain broad terms that will confuse the search (e.g. machine learning, algorithms, etc).
+I would like to use these keywords to retrieve code repositories related to these specific scientific disciplines from GitHub and Papers with Code.""",
+                },
+            ]
+
+            try:
+                result = await call_llm(
+                    messages,
+                    config["model"],
+                    config["base_url"],
+                    config["api_key"],
+                    pydantic_object=ExpandedKeywords,
+                    log_file=config["log_file"],
+                    temperature=0.5,
+                )
+                # Merge with base keywords and dedupe
+                expanded = list(set(config["keywords"] + result.get("keywords", [])))
+            except Exception as e:
+                logger.error(f"Keyword expansion failed: {e}, using base keywords")
+                expanded = config["keywords"]
+
+            # Save expanded keywords
+            import json
+
+            with open(config["keywords_expanded"], "w") as f:
+                json.dump({"keywords": expanded}, f, indent=2)
+            logger.info(f"[Done] Step 0: {len(expanded)} keywords: {expanded}")
+
+        # Step 1: Search GitHub repos with pending logic
+        logger.info("=" * 60 + "\nStep 1: Search GitHub Repos\n" + "=" * 60)
+        search_github_repos(expanded, config["github_token"], config["repos_searched"])
+
+        # Check final results
+        if config["repos_searched"].exists():
+            df_final = pd.read_csv(config["repos_searched"])
+            logger.info(f"[Done] Step 1: {len(df_final)} total repos in CSV")
+        else:
+            logger.warning("No repos found")
+            pd.DataFrame(
+                columns=[
+                    "keyword",
+                    "name",
+                    "full_name",
+                    "owner",
+                    "url",
+                    "description",
+                    "language",
+                    "topics",
+                    "stars",
+                    "forks",
+                    "created_at",
+                    "updated_at",
+                    "pushed_at",
+                    "license",
+                    "default_branch",
+                    "open_issues",
+                    "size",
+                    "has_wiki",
+                    "archived",
+                ]
+            ).to_csv(config["repos_searched"], index=False)
+            logger.info("[Done] Step 1: 0 repos saved")
+
+    # Step 2: Concurrent relevance check with resume support (batch processing)
+    # Step 3 & 4: Batch clone -> filter -> delete (runs in parallel with Step 2)
+    repos_check_history = config["workdir"] / "repos_check_history.csv"
+    
+    # Step 3 & 4 runner (supports one-shot / watch)
+    async def run_step34(*, watch: bool) -> None:
+        """Run Step 3 & 4: read YES from repos_check_history.csv, clone->filter->delete, with resume."""
+        logger.info("[步骤3&4] " + "=" * 60 + "\n[步骤3&4] Step 3 & 4: Background Processing (Clone -> Filter -> Delete)\n[步骤3&4] " + "=" * 60)
+        
+        # Setup directories
+        config["repos_raw"].mkdir(parents=True, exist_ok=True)
+        config["repos_filtered"].mkdir(parents=True, exist_ok=True)
+        
+        # Track processed repos
+        processed_urls = set()
+        processed_fullnames = set()  # Track by full_name for directory check
+        
+        # Check repos_processed.csv for already processed repos
+        if config["repos_processed"].exists():
+            df_processed = pd.read_csv(config["repos_processed"])
+            if not df_processed.empty:
+                processed_urls = set(df_processed["url"].tolist())
+                if "full_name" in df_processed.columns:
+                    processed_fullnames = set(df_processed["full_name"].tolist())
+                logger.info(f"[步骤3&4] 已处理记录: {len(processed_urls)} repos from repos_processed.csv")
+        
+        # Also check filtered directory for existing repos (in case CSV is missing but files exist)
+        if config["repos_filtered"].exists():
+            existing_dirs = [d.name for d in config["repos_filtered"].iterdir() if d.is_dir()]
+            existing_fullnames = {name.replace("___", "/") for name in existing_dirs}
+            processed_fullnames.update(existing_fullnames)
+            logger.info(f"[步骤3&4] 已存在目录: {len(existing_fullnames)} repos from repos_filtered directory")
+        
+        # Read all repos from repos_searched for mapping
+        if not config["repos_searched"].exists():
+            logger.error(f"[步骤3&4] 缺少 {config['repos_searched']}，无法将 url 映射回 repo 元数据；请先运行 step1/step2 或提供 repos_searched.csv")
+            return
+
+        df_all_repos = pd.read_csv(config["repos_searched"])
+        repo_map = {r["url"]: r for r in df_all_repos.to_dict("records")}
+        
+        poll_interval = int(args.poll_interval)
+        max_idle = int(args.max_idle)
+        consecutive_empty_checks = 0
+
+        async def process_once() -> int:
+            """Process pending YES repos once; return processed count in this round."""
+            nonlocal consecutive_empty_checks
+
+            if not repos_check_history.exists():
+                if watch and consecutive_empty_checks in {0, 4, 9, 19}:
+                    logger.info("[步骤3&4] 等待 repos_check_history.csv 文件生成...")
+                return 0
+
+            df_history = pd.read_csv(repos_check_history)
+            if df_history.empty:
+                if watch and consecutive_empty_checks in {0, 4, 9, 19}:
+                    logger.info("[步骤3&4] repos_check_history.csv 为空，等待新数据...")
+                return 0
+
+            df_relevant_history = df_history[df_history["is_relevant"] == "YES"].copy()
+            if df_relevant_history.empty:
+                if watch and consecutive_empty_checks in {0, 4, 9, 19}:
+                    logger.info("[步骤3&4] 暂无标记为 YES 的相关项目，等待新数据...")
+                return 0
+
+            relevant_urls = set(df_relevant_history["url"].tolist())
+            pending_urls = relevant_urls - processed_urls
+            if not pending_urls:
+                # nothing new to do
+                return 0
+
+            pending_repos: List[Dict[str, Any]] = []
+            for url in pending_urls:
+                if url in repo_map:
+                    repo = repo_map[url].copy()
+                    if repo.get("full_name") and repo["full_name"] not in processed_fullnames:
+                        pending_repos.append(repo)
+                else:
+                    history_record = df_relevant_history[df_relevant_history["url"] == url].iloc[0]
+                    full_name = history_record.get("full_name", "")
+                    repo = {
+                        "full_name": full_name,
+                        "url": url,
+                        "description": history_record.get("description", ""),
+                        "topics": history_record.get("topics", ""),
+                        "keyword": history_record.get("keyword", ""),
+                        "owner": full_name.split("/")[0] if "/" in full_name else "",
+                        "name": full_name.split("/")[1] if "/" in full_name else "",
+                    }
+                    if repo["full_name"] and repo["full_name"] not in processed_fullnames:
+                        pending_repos.append(repo)
+
+            if not pending_repos:
+                return 0
+
+            logger.info(
+                f"[步骤3&4] 📦 发现 {len(pending_repos)} 个新的相关项目需要处理（总共相关: {len(relevant_urls)}, 已处理: {len(processed_urls)}）"
+            )
+
+            processed_this_round = 0
+            total_batches = (len(pending_repos) + config["batch_size"] - 1) // config["batch_size"]
+            for batch_idx in range(total_batches):
+                start_idx = batch_idx * config["batch_size"]
+                end_idx = min(start_idx + config["batch_size"], len(pending_repos))
+                batch_repos = pending_repos[start_idx:end_idx]
+
+                logger.info(f"[步骤3&4] \n{'='*60}")
+                logger.info(f"[步骤3&4] Batch {batch_idx + 1}/{total_batches}: Processing {len(batch_repos)} repos")
+                logger.info(f"[步骤3&4] {'='*60}")
+
+                processed = await asyncio.to_thread(
+                    process_repos_batch,
+                    batch_repos,
+                    config["repos_raw"],
+                    config["repos_filtered"],
+                    config["repos_processed"],
+                )
+
+                for full_name in processed:
+                    processed_fullnames.add(full_name)
+                    for repo in batch_repos:
+                        if repo.get("full_name") == full_name:
+                            processed_urls.add(repo.get("url"))
+                            break
+
+                processed_this_round += len(processed)
+                logger.info(
+                    f"[步骤3&4] ✓ Batch {batch_idx + 1}/{total_batches}: {len(processed)}/{len(batch_repos)} repos processed successfully"
+                )
+
+            return processed_this_round
+
+        if not watch:
+            try:
+                n = await process_once()
+                if n == 0:
+                    logger.info("[步骤3&4] 本轮无新项目可处理，退出（未开启 --watch）")
+                else:
+                    logger.info(f"[步骤3&4] 本轮处理完成：新增处理 {n} 个 repo，退出（未开启 --watch）")
+            except Exception as e:
+                logger.error(f"[步骤3&4] 处理相关项目时出错: {e}")
+            return
+
+        # watch loop
+        while True:
+            try:
+                n = await process_once()
+                if n > 0:
+                    consecutive_empty_checks = 0
+                else:
+                    consecutive_empty_checks += 1
+                    if consecutive_empty_checks >= max_idle:
+                        logger.info(f"[步骤3&4] 连续空转 {consecutive_empty_checks} 次，退出（watch 模式）")
+                        break
+                await asyncio.sleep(poll_interval)
+            except Exception as e:
+                consecutive_empty_checks += 1
+                logger.error(f"[步骤3&4] 处理相关项目时出错: {e}")
+                if consecutive_empty_checks >= max_idle:
+                    logger.error("[步骤3&4] 连续错误次数过多，退出（watch 模式）")
+                    break
+                await asyncio.sleep(poll_interval)
+
+        logger.info(f"[步骤3&4] [Done] 步骤3&4退出：已处理 {len(processed_urls)} 个相关项目")
+    
+    background_task = None
+    if args.mode in {"all"}:
+        # Start background task for Step 3 & 4 (will run in parallel with Step 2)
+        background_task = asyncio.create_task(run_step34(watch=True))
+        logger.info("✓ 已启动后台任务：步骤3&4（克隆和过滤）将与步骤2并行运行")
+    elif args.mode == "step34":
+        # step34 only
+        await run_step34(watch=bool(args.watch))
+        return
+    
+    if args.mode == "step2":
+        logger.info("[步骤2] " + "=" * 60 + "\n[步骤2] Step 2: Check Relevance with LLM (Batch Concurrent)\n[步骤2] " + "=" * 60)
+    else:
+        # args.mode == "all"
+        logger.info("[步骤2] " + "=" * 60 + "\n[步骤2] Step 2: Check Relevance with LLM (Batch Concurrent)\n[步骤2] " + "=" * 60)
+
+    # read to check list
+    df_to_check = pd.read_csv(config["repos_searched"])
+    total_repos = len(df_to_check)
+    to_check_urls = set(df_to_check["url"].tolist())
+
+    # read already checked list
+    already_checked_urls = set()
+    if repos_check_history.exists():
+        df_checked = pd.read_csv(repos_check_history)
+        already_checked_urls = set(df_checked["url"].tolist())
+        logger.info(f"[步骤2] Resume: Already checked {len(already_checked_urls)}/{total_repos} repos")
+
+    # pending repos to check
+    pending_urls = to_check_urls - already_checked_urls
+    unchecked = df_to_check[df_to_check["url"].isin(pending_urls)].to_dict("records")
+
+    if not unchecked:
+        logger.info(f"[步骤2] [Skip] Step 2: All {total_repos} repos have been checked")
+        # Still need to generate repos_checked.csv if it doesn't exist
+        if not config["repos_checked"].exists():
+            # collect all relevant repos from history file
+            all_repos = df_to_check.to_dict("records")
+            relevant_repos = []
+            if repos_check_history.exists():
+                df_history = pd.read_csv(repos_check_history)
+                if not df_history.empty:
+                    relevant_urls = set(df_history[df_history["is_relevant"] == "YES"]["url"].tolist())
+                    relevant_repos = [r for r in all_repos if r["url"] in relevant_urls]
+
+            # deduplicate by url and save
+            if relevant_repos:
+                df_relevant = pd.DataFrame(relevant_repos)
+                df_relevant = df_relevant.drop_duplicates(subset=["url"])
+                df_relevant.to_csv(config["repos_checked"], index=False, encoding="utf-8")
+                logger.info(f"[步骤2] [Done] Step 2: {len(df_relevant)} relevant repos (deduplicated)")
+            else:
+                pd.DataFrame(columns=df_to_check.columns).to_csv(config["repos_checked"], index=False)
+                logger.info("[步骤2] [Done] Step 2: 0 relevant repos")
+    else:
+        logger.info(f"[步骤2] Pending: {len(unchecked)} repos to check")
+        # create parser
+        parser = JsonOutputParser(pydantic_object=RelevanceResult)
+        format_instructions = parser.get_format_instructions()
+        keywords_str = ", ".join(expanded)
+        system_content = f"""You are an expert at reading GitHub README.md files thoroughly and determining whether the repository hosts scientific code that is relevant to the scientific disciplines of {keywords_str}.
+Your task is to decide if the repository's scientific code is related to these disciplines.
+Only answer based on the information available in the README, repository description, and topics.
+{format_instructions}"""
+
+        # Check if using reasoning model (e.g., Qwen3-32B-AWQ)
+        is_reasoning_model = "qwen" in config["model"].lower() or "reasoning" in config["model"].lower()
+        
+        # Get base URLs for parallel processing
+        base_urls = config["base_urls"]
+        num_gpus = len(base_urls)
+        logger.info(f"[步骤2] Using {num_gpus} VLLM service(s): {base_urls}")
+        
+        # concurrent check function with base_url parameter
+        async def check_one(repo, base_url: str):
+            repo_name = repo['full_name']
+            logger.info(f"[步骤2] 🔄 开始检查: {repo_name}")
+            try:
+                readme = (await get_readme(repo["owner"], repo["name"], config["github_token"]))[:8000]
+                logger.info(f"[步骤2] ✓ 已获取 README: {repo_name}, 长度: {len(readme)}")
+            except Exception as e:
+                logger.error(f"[步骤2] ✗ 获取 README 失败: {repo_name}, 错误: {e}")
+                return None
+            messages = [
+                {"role": "system", "content": system_content},
+                {
+                    "role": "user",
+                    "content": f"""Think before you respond. Your answer should be based on your thorough understanding of the content of the README.md file.
+Does the README.md file indicate that the repository hosts code related to the scientific disciplines of {keywords_str}?
+Repository: {repo['full_name']}
+Description: {repo['description']}
+Topics: {repo['topics']}
+README: {readme}
+Answer by 'YES' or 'NO' in the relevant field. And provide your reasoning in the reason field.""",
+                },
+            ]
+            try:
+                logger.info(f"[步骤2] 🤖 调用 LLM: {repo_name} (base_url: {base_url})")
+                result = await call_llm(
+                    messages,
+                    config["model"],
+                    base_url,  # Use provided base_url
+                    config["api_key"],
+                    pydantic_object=RelevanceResult,
+                    log_file=config["log_file"],
+                    temperature=0.1,
+                    max_tokens=config["max_tokens"],  # Limit generation length
+                )
+                
+                # 检查 LLM 调用是否成功
+                if result is None:
+                    logger.error(f"[步骤2] ❌ LLM 调用失败（超时或错误）: {repo_name}")
+                    return None
+                
+                logger.info(f"[步骤2] ✓ LLM 调用完成: {repo_name}")
+                
+                # Additional post-processing for reasoning models if needed
+                # (call_llm already handles most cases, but this is a safety check)
+                if is_reasoning_model:
+                    if not isinstance(result, dict) or "relevant" not in result:
+                        logger.warning(f"[步骤2] 推理模型响应格式异常，尝试后处理: {repo['full_name']}")
+                        # If result is a string, extract from it
+                        if isinstance(result, str):
+                            result = extract_final_answer_from_reasoning(result, RelevanceResult)
+                        # If result dict is malformed, try to extract from reason field
+                        elif isinstance(result, dict):
+                            raw_text = result.get("reason", str(result))
+                            result = extract_final_answer_from_reasoning(raw_text, RelevanceResult)
+                
+                is_relevant = result.get("relevant", "").upper() == "YES"
+                reason = result.get("reason", "")
+
+                # return result for batch write
+                return {
+                    "keyword": repo["keyword"],
+                    "full_name": repo["full_name"],
+                    "url": repo["url"],
+                    "description": repo["description"],
+                    "topics": repo["topics"],
+                    "is_relevant": "YES" if is_relevant else "NO",
+                    "reason": reason,
+                    "relevant": is_relevant,
+                }
+
+            except Exception as e:
+                logger.error(f"[步骤2] Error checking {repo['full_name']}: {e}")
+                return None
+
+        # batch processing
+        BATCH_SIZE = 20
+        total_batches = (len(unchecked) + BATCH_SIZE - 1) // BATCH_SIZE
+
+        for batch_idx in range(total_batches):
+            start_idx = batch_idx * BATCH_SIZE
+            end_idx = min(start_idx + BATCH_SIZE, len(unchecked))
+            batch = unchecked[start_idx:end_idx]
+
+            logger.info(f"[步骤2] \n{'='*60}")
+            logger.info(f"[步骤2] 📦 Batch {batch_idx + 1}/{total_batches}: Processing {len(batch)} repos")
+            logger.info(f"[步骤2] {'='*60}")
+
+            # Split batch across multiple GPUs for parallel processing
+            if num_gpus > 1:
+                # Divide batch into sub-batches for each GPU
+                sub_batch_size = len(batch) // num_gpus
+                sub_batches = []
+                for gpu_idx in range(num_gpus):
+                    sub_start = gpu_idx * sub_batch_size
+                    if gpu_idx == num_gpus - 1:
+                        # Last GPU gets remaining items
+                        sub_batch = batch[sub_start:]
+                    else:
+                        sub_batch = batch[sub_start:sub_start + sub_batch_size]
+                    if sub_batch:  # Only add non-empty batches
+                        sub_batches.append((sub_batch, base_urls[gpu_idx], gpu_idx))
+                        logger.info(f"[步骤2]   GPU {gpu_idx + 1} ({base_urls[gpu_idx]}): {len(sub_batch)} repos")
+                
+                # Process sub-batches in parallel across GPUs
+                gpu_tasks = []
+                for sub_batch, base_url, gpu_idx in sub_batches:
+                    tasks = [check_one(repo, base_url) for repo in sub_batch]
+                    gpu_tasks.append(asyncio.gather(*tasks))
+                
+                # Execute all GPU tasks in parallel with timeout protection
+                if gpu_tasks:
+                    logger.info(f"[步骤2] ⏳ 等待 {len(gpu_tasks)} 个 GPU 任务完成...")
+                    try:
+                        # 为整个 batch 添加超时保护（每个任务最多 120 秒，batch 最多等待 5 分钟）
+                        batch_timeout = max(300, len(batch) * 2)  # 至少 5 分钟，或每个任务 2 分钟
+                        gpu_results = await asyncio.wait_for(
+                            asyncio.gather(*gpu_tasks, return_exceptions=True),
+                            timeout=batch_timeout
+                        )
+                        # Flatten results from all GPUs
+                        batch_results = []
+                        for sublist in gpu_results:
+                            if isinstance(sublist, Exception):
+                                logger.error(f"[步骤2] GPU 任务异常: {sublist}")
+                                batch_results.append(None)
+                            else:
+                                batch_results.extend(sublist)
+                    except asyncio.TimeoutError:
+                        logger.error(f"[步骤2] ❌ Batch {batch_idx + 1} 超时（{batch_timeout}秒），跳过剩余任务")
+                        batch_results = [None] * len(batch)  # 标记为失败
+                else:
+                    batch_results = []
+            else:
+                # Single GPU: use original logic with timeout
+                logger.info(f"[步骤2] ⏳ 等待 {len(batch)} 个任务完成...")
+                try:
+                    batch_timeout = max(300, len(batch) * 2)
+                    batch_results = await asyncio.wait_for(
+                        asyncio.gather(*[check_one(r, base_urls[0]) for r in batch], return_exceptions=True),
+                        timeout=batch_timeout
+                    )
+                    # 处理异常结果
+                    batch_results = [r if not isinstance(r, Exception) else None for r in batch_results]
+                except asyncio.TimeoutError:
+                    logger.error(f"[步骤2] ❌ Batch {batch_idx + 1} 超时（{batch_timeout}秒），跳过剩余任务")
+                    batch_results = [None] * len(batch)
+
+            # batch write result to history file
+            valid_results = [r for r in batch_results if r is not None]
+            if valid_results:
+                df_batch = pd.DataFrame(valid_results)
+                # only keep columns to write to CSV
+                df_batch = df_batch[
+                    ["keyword", "full_name", "url", "description", "topics", "is_relevant", "reason"]
+                ]
+                df_batch.to_csv(
+                    repos_check_history,
+                    mode="a",
+                    header=not repos_check_history.exists(),
+                    index=False,
+                    encoding="utf-8",
+                )
+
+                for result in valid_results:
+                    status = "✓ Relevant" if result["relevant"] else "✗ Not relevant"
+                    logger.info(f"[步骤2]   {status}: {result['full_name']}")
+
+                logger.info(f"[步骤2] ✓ Batch {batch_idx + 1}/{total_batches}: {len(valid_results)} repos saved")
+
+        # collect all relevant repos from history file
+        all_repos = df_to_check.to_dict("records")
+        relevant_repos = []
+        if repos_check_history.exists():
+            df_history = pd.read_csv(repos_check_history)
+            if not df_history.empty:
+                relevant_urls = set(df_history[df_history["is_relevant"] == "YES"]["url"].tolist())
+                relevant_repos = [r for r in all_repos if r["url"] in relevant_urls]
+
+        # deduplicate by url and save
+        if relevant_repos:
+            df_relevant = pd.DataFrame(relevant_repos)
+            df_relevant = df_relevant.drop_duplicates(subset=["url"])
+            df_relevant.to_csv(config["repos_checked"], index=False, encoding="utf-8")
+            logger.info(f"[步骤2] [Done] Step 2: {len(df_relevant)} relevant repos (deduplicated)")
+        else:
+            pd.DataFrame(columns=df_to_check.columns).to_csv(config["repos_checked"], index=False)
+            logger.info("[步骤2] [Done] Step 2: 0 relevant repos")
+    
+    if args.mode == "all" and background_task is not None:
+        # Step 2 completed, wait for background task to process remaining repos
+        logger.info("[步骤2] 步骤2已完成，等待步骤3&4后台任务处理所有相关项目...")
+        try:
+            await asyncio.wait_for(asyncio.shield(background_task), timeout=600)
+            logger.info("[步骤3&4] ✓ 步骤3&4后台任务已完成")
+        except asyncio.TimeoutError:
+            logger.info("[步骤3&4] ⏱️ 等待步骤3&4超过 600 秒，步骤3&4将继续在后台处理（未被取消）...")
+            logger.info("[步骤3&4]    提示：下次运行时，程序会自动检查已处理的项目，只处理新的项目")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())