data1/main_v2.py

# 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())