#!/usr/bin/env python3
"""Additive full-catalog backfill and asset continuation. Never deletes user data."""
import argparse
import csv
import fcntl
import hashlib
import json
import os
import re
import shutil
import sqlite3
import tarfile
import time
import uuid
import html
import contextlib
import signal
from collections import Counter
from pathlib import Path
from urllib.parse import urljoin, urlsplit, urlunsplit
import requests
from bs4 import BeautifulSoup
import download_competitions as old
csv.field_size_limit(100*1024**2)

ROOT=old.ROOT
STATE=ROOT/'download_state'/'full-v2'
RESERVE=100*1024**3
VERSION=2
MODEL_CACHE={}
SOURCE_CACHE={}
METADATA_ACTIVE=None
METADATA_JOB_SECONDS=300
METADATA_ATTACHMENT_SECONDS=90
RECEIPT_DB=Path('/var/lib/dataset-move/handoff.sqlite')

class MetadataTimeout(BaseException):pass
class AttachmentTimeout(Exception):pass

def install_metadata_transport():
    original=requests.sessions.Session.request
    if getattr(original,'metadata_bounded',False):return
    def bounded(self,method,url,**kwargs):
        if kwargs.get('timeout') is None:kwargs['timeout']=(15,75)
        return original(self,method,url,**kwargs)
    bounded.metadata_bounded=True
    requests.sessions.Session.request=bounded

def arm_metadata_timer():
    if METADATA_ACTIVE is None:return
    current=time.monotonic();a=METADATA_ACTIVE
    delays=[a['heartbeat'],a['deadline']-current]
    if a.get('attachment_deadline') is not None:delays.append(a['attachment_deadline']-current)
    signal.setitimer(signal.ITIMER_REAL,max(.001,min(delays)))

@contextlib.contextmanager
def metadata_deadline(progress,seconds=None,heartbeat=15):
    global METADATA_ACTIVE
    previous=METADATA_ACTIVE;handler=signal.getsignal(signal.SIGALRM);timer=signal.getitimer(signal.ITIMER_REAL)
    started=time.monotonic();limit=METADATA_JOB_SECONDS if seconds is None else seconds
    METADATA_ACTIVE=dict(deadline=started+limit,heartbeat=heartbeat,progress=progress)
    def tick(signum,frame):
        current=time.monotonic();a=METADATA_ACTIVE
        if current>=a['deadline']:raise MetadataTimeout('competition exceeded metadata time budget; saved files retained')
        if a.get('attachment_deadline') is not None and current>=a['attachment_deadline']:
            raise AttachmentTimeout('attachment exceeded wall-clock budget; partial retained for later range resume')
        status('metadata','live_backfill',**progress,heartbeat_at=now(),elapsed_seconds=round(current-started,1),operation=a.get('operation','metadata'))
        arm_metadata_timer()
    signal.signal(signal.SIGALRM,tick);arm_metadata_timer()
    try:yield
    finally:
        signal.setitimer(signal.ITIMER_REAL,0);signal.signal(signal.SIGALRM,handler);METADATA_ACTIVE=previous
        if previous is not None:arm_metadata_timer()
        elif timer[0]:signal.setitimer(signal.ITIMER_REAL,max(.001,timer[0]-(time.monotonic()-started)),timer[1])

@contextlib.contextmanager
def attachment_deadline():
    if METADATA_ACTIVE is None:yield;return
    a=METADATA_ACTIVE;previous=(a.get('attachment_deadline'),a.get('operation'))
    a.update(attachment_deadline=time.monotonic()+METADATA_ATTACHMENT_SECONDS,operation='page_attachment');arm_metadata_timer()
    try:yield
    finally:
        a['attachment_deadline'],a['operation']=previous;arm_metadata_timer()

def delivered_attachment(path):
    if path.exists() or not RECEIPT_DB.exists():return None
    relative=str(path.relative_to(ROOT))
    with sqlite3.connect('file:'+str(RECEIPT_DB)+'?mode=ro',uri=True) as db:
        hit=db.execute("SELECT id,bytes FROM receipts WHERE dataset='kaggle' AND path=? AND status='removed' AND note='verified_and_unlinked' LIMIT 1",(relative,)).fetchone()
    if hit:return dict(status='delivered_external',receipt_id=hit[0],bytes=hit[1],local_path=str(path.relative_to(old.OUT_ROOT)))

def catalog_digest(rows):
    return hashlib.sha256(json.dumps([(r['id'],r['slug']) for r in rows]).encode()).hexdigest()

def metadata_resume_index(rows,previous,checkpoint):
    if checkpoint.get('catalog_digest')==catalog_digest(rows):
        value=checkpoint.get('next_index',0)
        if isinstance(value,int) and 0<=value<len(rows):return value
    slug=previous.get('current_slug') if previous.get('phase')=='live_backfill' else None
    return next((i for i,r in enumerate(rows) if r['slug']==slug),0)

@contextlib.contextmanager
def metadata_competition_lock(slug):
    path=Path('/var/lib/dataset-move/competition-locks')/(slug+'.lock');path.parent.mkdir(parents=True,exist_ok=True)
    with path.open('a') as lease:
        fcntl.flock(lease,fcntl.LOCK_EX);yield
def now():return time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
def read(p,default=None):
    try:return json.loads(Path(p).read_text())
    except (OSError,ValueError):return default
def atomic(p,x):
    p=Path(p);p.parent.mkdir(parents=True,exist_ok=True)
    tmp=p.with_name(p.name+'.tmp-'+uuid.uuid4().hex)
    tmp.write_text(json.dumps(x,ensure_ascii=False,indent=2) if not isinstance(x,str) else x)
    tmp.chmod(0o644);os.replace(tmp,p)
def room(heavy=False,need=0):
    d=shutil.disk_usage(ROOT)
    return d.free-need>RESERVE and (not heavy or old.df_used_percent(d.used,d.free)<90)
def status(worker,phase,**kw):
    d=shutil.disk_usage(ROOT)
    atomic(STATE/(worker+'-status.json'),dict(state='running',phase=phase,updated_at=now(),pid=os.getpid(),free_bytes=d.free,**kw))
def wait_room(worker,heavy=False):
    while not room(heavy):
        status(worker,'waiting_for_space',reserve_bytes=RESERVE,heavy_resume_below_percent=90 if heavy else None)
        time.sleep(60)
def source_ok(p):
    if not p.is_file() or p.stat().st_size==0:return False
    if p.suffix.lower()=='.ipynb':
        x=read(p);return isinstance(x,dict) and isinstance(x.get('cells'),list)
    with p.open('rb') as f:head=f.read(256).lower()
    return p.suffix.lower() in ('.py','.r','.rmd','.sql','.jl','.src','.irnb','.ijlnb') and b'<html' not in head
def sources(folder):return [p for p in folder.glob('*') if source_ok(p)]
def gzip_ok(p):
    try:
        with tarfile.open(p,'r:gz') as t:return t.next() is not None
    except (OSError,tarfile.TarError,EOFError):return False
def safe_sdk_download(self,response,outfile,http_client,quiet=True,resume=False,**kwargs):
    """Stream only into repair-owned staging, with identity-checked range resume."""
    p=Path(outfile)
    if '.repair-download' not in p.parts:raise RuntimeError('SDK writes must use repair staging')
    p.parent.mkdir(parents=True,exist_ok=True)
    encoded=response.headers.get('Content-Encoding','identity').lower() not in ('','identity')
    # iter_content decodes HTTP compression; wire length/ranges no longer match it.
    total=0 if encoded else int(response.headers.get('Content-Length') or 0)
    validator=None if encoded else (response.headers.get('ETag') or response.headers.get('Last-Modified'))
    marker=p.with_name(p.name+'.resume.json');saved=read(marker,{})
    heavy=kwargs.get('heavy',True)
    offset=p.stat().st_size if p.exists() else 0
    if p.exists() and offset==0:p.rename(p.with_name(p.name+'.preserved-'+uuid.uuid4().hex))
    if offset and (not validator or saved.get('validator')!=validator or saved.get('size')!=total):
        p.rename(p.with_name(p.name+'.preserved-'+uuid.uuid4().hex));offset=0
    if total and offset==total:return
    if not room(heavy,max(0,total-offset)):raise old.DiskLimit('download does not fit disk reserve')
    if offset:
        current=requests.get(response.url,headers={'Range':f'bytes={offset}-','If-Range':validator},stream=True,timeout=(20,120))
        if current.status_code!=206 or not current.headers.get('Content-Range','').startswith(f'bytes {offset}-'):
            current.close();raise old.Incomplete('range resume was not confirmed; preserved partial')
        response.close();response=current
    if hasattr(response,'raise_for_status'):response.raise_for_status()
    atomic(marker,dict(validator=validator,size=total))
    try:
        chunks=response.iter_content(1024**2) if hasattr(response,'iter_content') else iter(lambda:response.read(1024**2),b'')
        with p.open('ab' if offset else 'xb') as out:
            for chunk in chunks:
                if not chunk:continue
                if not room(heavy,len(chunk)):raise old.DiskLimit('disk reserve reached during stream')
                out.write(chunk)
        if total and p.stat().st_size!=total:raise old.Incomplete('download length mismatch')
    finally:response.close()
def safe_pull(api,ref,dest,version_id=None):
    dest.mkdir(parents=True,exist_ok=True)
    if sources(dest):return
    cached=SOURCE_CACHE.get(ref)
    if cached and sources(cached):
        for p in sources(cached):
            target=dest/p.name
            if not target.exists():os.link(p,target)
        atomic(dest/'.ok',ref+'\n');return
    src=old.local_kernel_file(version_id)
    if src and source_ok(src):
        target=dest/(old.safe_name(ref.split('/',1)[1])+src.suffix)
        if target.exists():target=dest/('repair-'+str(version_id)+'-'+target.name)
        if target.exists() and not source_ok(target):target=dest/('repair-'+uuid.uuid4().hex+src.suffix)
        if not target.exists():os.link(src,target)
        atomic(dest/'.ok',ref+'\n');SOURCE_CACHE[ref]=dest;return
    old.ensure_disk_room('kernel '+ref)
    stage=dest/'.repair-download';stage.mkdir(exist_ok=True)
    if not sources(stage):old.call_sdk(lambda:api.kernels_pull(ref,str(stage),metadata=True,quiet=True),'pull '+ref)
    found=sources(stage)
    if not found:raise old.Incomplete('kernel source missing or invalid')
    for p in found:
        target=dest/p.name
        if target.exists():target=dest/('repair-'+p.name)
        if not target.exists():os.link(p,target)
    atomic(dest/'.ok',ref+'\n');SOURCE_CACHE[ref]=dest
def safe_model(api,ref,dest):
    dest.mkdir(parents=True,exist_ok=True)
    marker=dest/'.ok'
    records=read(dest/'.model-files.json',{})
    item=records.get(ref)
    if item:
        p=dest/item['file']
        if p.is_file() and p.stat().st_size==item['bytes'] and gzip_ok(p):return p
    def remember(p):
        records[ref]=dict(file=p.name,bytes=p.stat().st_size,verified_at=now())
        atomic(dest/'.model-files.json',records);atomic(marker,ref+'\n');MODEL_CACHE[ref]=p
        return p
    cached=MODEL_CACHE.get(ref)
    if cached and cached.is_file() and gzip_ok(cached):
        target=dest/cached.name
        if target.exists() and not os.path.samefile(target,cached):target=dest/('repair-'+uuid.uuid4().hex+'-'+cached.name)
        if not target.exists():os.link(cached,target)
        if gzip_ok(target):return remember(target)
    if marker.exists() and marker.read_text().strip()==ref:
        archives=list(dest.glob('*.tar.gz'))
        if len(archives)==1 and gzip_ok(archives[0]):return remember(archives[0])
    old.ensure_disk_room('model '+ref)
    stage=dest/'.repair-download'/('version-'+ref.rsplit('/',1)[1]);stage.mkdir(parents=True,exist_ok=True)
    path=Path(old.call_sdk(lambda:api.model_instance_version_download(ref,path=str(stage),force=False,quiet=True,untar=False),'model '+ref))
    if not gzip_ok(path):raise old.Incomplete('model archive failed tar/gzip validation')
    target=dest/path.name
    if target.exists():target=dest/('repair-v'+ref.rsplit('/',1)[1]+'-'+path.name)
    if not target.exists():os.link(path,target)
    return remember(target)
def install_safety(api):
    old.pull_kernel=safe_pull;old.download_model=safe_model;old.write_json=legacy_write
    from types import MethodType
    api.download_file=MethodType(safe_sdk_download,api)

def legacy_write(path,payload):
    path=Path(path)
    if path.exists():
        previous=path.read_bytes()
        if read(path)==payload:return
        key=hashlib.sha256(previous).hexdigest()
        backup=STATE/'history'/str(path.relative_to(ROOT))/(key+'.previous')
        if not backup.exists():backup.parent.mkdir(parents=True,exist_ok=True);shutil.copy2(path,backup)
    atomic(path,payload)

class RemoteError(Exception):pass
def request(sess,method,url,payload=None):
    last=''
    for attempt in range(4):
        time.sleep(.35)
        try:
            r=sess.request(method,url,json=payload,timeout=(15,75))
            if r.status_code in (429,500,502,503,504):
                last=f'HTTP {r.status_code}';time.sleep(min(60,float(r.headers.get('Retry-After') or 2**(attempt+1))));continue
            if r.status_code!=200:raise RemoteError(f'HTTP {r.status_code}')
            x=r.json()
            if not isinstance(x,dict) or x.get('error'):raise RemoteError('unexpected API body')
            return x
        except (requests.RequestException,ValueError) as e:last=type(e).__name__
    raise RemoteError(last or 'API retries exhausted')

def build_vote_db():
    p=STATE/'votes.sqlite';db=sqlite3.connect(p)
    db.execute('CREATE TABLE IF NOT EXISTS votes(id TEXT PRIMARY KEY,votes INTEGER,version TEXT)')
    ready=read(STATE/'votes-ready.json',{})
    src=old.META_DIR/'Kernels.csv'
    if ready.get('source_mtime')==src.stat().st_mtime_ns:return db
    with src.open(newline='') as f:
        batch=[]
        for i,row in enumerate(csv.DictReader(f),1):
            value=row.get('TotalVotes','').strip()
            batch.append((row['Id'],int(value) if value else None,row.get('CurrentKernelVersionId')))
            if len(batch)==10000:
                db.executemany('INSERT OR REPLACE INTO votes VALUES(?,?,?)',batch);db.commit();batch=[]
                if i%1000000==0:status('metadata','indexing_votes',rows=i)
        if batch:db.executemany('INSERT OR REPLACE INTO votes VALUES(?,?,?)',batch);db.commit()
    atomic(STATE/'votes-ready.json',dict(source_mtime=src.stat().st_mtime_ns,source_date=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime(src.stat().st_mtime))))
    return db
def metadata_rows():
    wanted={str(r['id']) for r in old.load_competitions(old.CSV_PATH)}
    with (old.META_DIR/'Competitions.csv').open(newline='') as f:
        return {r['Id']:r for r in csv.DictReader(f) if r['Id'] in wanted}
def bootstrap(rows,meta,db):
    index=old.cached_code_index() or {}
    for i,row in enumerate(rows,1):
        wait_room('metadata')
        dest=old.OUT_ROOT/row['slug'];m=meta.get(str(row['id']),{})
        desc=m.get('DatasetDescription','')
        path=dest/'data'/'overview-meta.md'
        if desc and not path.exists():
            atomic(path,'# '+row['title']+' — Data\n\n来源：'+row['url']+'/data\n\n本地 Meta Kaggle 快照；不是本次实时页面。\n\n'+desc)
        votes=[]
        for k in index.get(str(row['id']),[]):
            hit=db.execute('SELECT votes FROM votes WHERE id=?',(str(k.get('kernel_id')),)).fetchone()
            votes.append(dict(ref=k['ref'],kernel_id=k.get('kernel_id'),votes=hit[0] if hit else None))
        vp=dest/'code'/'votes-meta.json'
        if not vp.exists():atomic(vp,dict(source='Meta Kaggle Kernels.csv',source_snapshot=read(STATE/'votes-ready.json'),items=votes,count=len(votes)))
        if i%100==0:status('metadata','offline_backfill',done=i,total=len(rows))
    status('metadata','offline_backfill_complete',done=len(rows),total=len(rows))

def markdown(content):
    if not re.search(r'<(?:p|div|h[1-6]|table|br)\b',content,re.I):return content
    soup=BeautifulSoup(content,'html.parser')
    for a in soup.find_all('a',href=True):a.replace_with('['+a.get_text(' ',strip=True)+']('+a['href']+')')
    for im in soup.find_all('img',src=True):im.replace_with('!['+im.get('alt','')+']('+im['src']+')')
    return soup.get_text('\n',strip=True)

def asset_url(url):
    url=html.unescape(html.unescape(url))
    u=urlsplit(url)
    if u.hostname=='github.com' and '/blob/' in u.path:
        u=u._replace(netloc='raw.githubusercontent.com',path=u.path.replace('/blob/','/',1))
    if u.hostname=='storage.cloud.google.com':u=u._replace(netloc='storage.googleapis.com')
    return urlunsplit(u._replace(fragment=''))
def local_markdown(content,assets,folder):
    text=markdown(content)
    for asset in assets:
        if asset.get('status')!='ok':continue
        remote=asset['url'];local=os.path.relpath(old.OUT_ROOT/asset['local_path'],folder)
        text=text.replace(remote,local).replace(html.escape(remote,quote=False),local)
    return text

def page_assets(content,base,folder):
    """Collect images and file attachments explicitly referenced by the page."""
    targets=set(re.findall(r'!\[[^\]]*\]\((https?://[^\s)]+)',content))
    soup=BeautifulSoup(content,'html.parser')
    targets.update(urljoin(base,x['src']) for x in soup.find_all('img',src=True))
    for u in re.findall(r'https?://[^\s<>"\)]+',content):
        if Path(urlsplit(u).path).suffix.lower() in ('.png','.jpg','.jpeg','.webp','.gif','.svg','.pdf','.zip','.ipynb','.py','.tar','.gz'):targets.add(u)
    result=[]
    for url in sorted({asset_url(u) for u in targets}):
        parsed=urlsplit(url)
        if parsed.scheme!='https' or not parsed.hostname or parsed.username or parsed.password:
            result.append(dict(url=url,status='unsupported_url'));continue
        suffix=Path(parsed.path).suffix.lower()
        if not re.fullmatch(r'\.[a-z0-9]{1,8}',suffix or ''):suffix='.bin'
        key=hashlib.sha256(url.encode()).hexdigest()[:24]+suffix
        final=folder/key
        receipt=delivered_attachment(final)
        if receipt:
            result.append(dict(receipt,url=url));continue
        done=read(folder/(key+'.json'),{})
        if done.get('status')=='ok' and final.is_file() and final.stat().st_size==done.get('bytes'):
            result.append(done);continue
        try:
            # No Kaggle authentication is sent to external attachment hosts.
            stage=folder/'.repair-download'/key
            with attachment_deadline():
                with requests.get(url,headers={'Accept-Encoding':'identity'},stream=True,timeout=(15,75)) as response:
                    response.raise_for_status()
                    safe_sdk_download(None,response,str(stage),None,heavy=False)
            with stage.open('rb') as f:head=f.read(1024).lower()
            if b'<html' in head or b'<!doctype html' in head:raise ValueError('attachment returned HTML')
            if suffix=='.pdf' and b'%pdf-' not in head:raise ValueError('not a PDF')
            if suffix=='.ipynb' and not source_ok(stage):raise ValueError('not a notebook')
            if final.exists():final=folder/('verified-'+key)
            if not final.exists():os.link(stage,final)
            rec=dict(url=url,status='ok',local_path=str(final.relative_to(old.OUT_ROOT)),bytes=final.stat().st_size,fetched_at=now())
            atomic(folder/(key+'.json'),rec);result.append(rec)
        except AttachmentTimeout as e:result.append(dict(url=url,status='deferred_timeout',error=str(e),partial_preserved=True))
        except old.DiskLimit:result.append(dict(url=url,status='waiting_for_space'))
        except Exception as e:result.append(dict(url=url,status='failed',error=type(e).__name__+': '+str(e)[:150]))
    return result
def normalize_board(data,expected=None):
    result={};teams={str(t['teamId']):t for t in data.get('teams',[]) if t.get('teamId')}
    for key in ('publicLeaderboard','privateLeaderboard'):
        rows=[];seen=set()
        for raw in data.get(key,[]):
            tid=str(raw.get('teamId'))
            if tid in seen:continue
            seen.add(tid);team=teams.get(tid,{})
            rows.append(dict(team_id=raw.get('teamId'),team_name=team.get('teamName'),rank=raw.get('rank'),score=raw.get('displayScore',raw.get('score')),medal=raw.get('medal'),solution_url=team.get('solutionWriteUpUrl'),solution_topic_id=team.get('writeUpForumTopicId'),raw=raw,team=team))
        result[key]=rows
    n=max((len(v) for v in result.values()),default=0)
    result.update(status='ok' if (expected is not None and n>=expected and not data.get('nextPageToken')) else 'partial',expected_teams=expected,actual_teams=n)
    return result
def solution_thread(api,sess,row,team,dest):
    tid=team.get('writeUpForumTopicId');url=team.get('solutionWriteUpUrl')
    if not tid:return dict(url=url,status='missing_topic_id')
    folder=dest/'leaderboard'/'solutions'/str(tid);saved=read(folder/'thread.json')
    if saved and saved.get('complete'):
        assets=page_assets(saved.get('content',''),row['url'],folder/'assets')
        atomic(folder/'solution.md','# '+saved.get('title','Solution')+'\n\n'+local_markdown(saved.get('content',''),assets,folder))
        return dict(topic_id=tid,url=url,status='ok' if all(a['status'] in ('ok','delivered_external') for a in assets) else 'partial',local_path=str((folder/'thread.json').relative_to(dest)),reused=True,assets=assets)
    existing=read(dest/'discussion'/(str(tid)+'.json'))
    if existing and existing.get('content') and old.count_comments(existing.get('comments',[]))>=int(existing.get('comment_count') or 0):
        payload=dict(existing);payload.update(complete=True,source='existing_discussion')
    else:
        topic,comments=old.fetch_comment_tree(api,int(tid))
        if not topic:raise old.Incomplete('solution topic missing')
        msgs=old.raw_messages(sess,row['slug'],int(tid))
        payload=dict(id=int(tid),title=topic.title,content=topic.content,comments=[old.comment_to_dict(c) for c in comments],comment_count=topic.comment_count,messages=msgs,complete=True,source='live',fetched_at=now())
    payload.update(competition_id=int(row['id']),competition_slug=row['slug'],team_id=team.get('teamId'),solution_url=url)
    atomic(folder/'thread.json',payload)
    body=payload.get('content','')
    assets=page_assets(body,row['url'],folder/'assets')
    atomic(folder/'solution.md','# '+payload.get('title','Solution')+'\n\n'+local_markdown(body,assets,folder))
    return dict(topic_id=tid,url=url,status='ok' if all(a['status'] in ('ok','delivered_external') for a in assets) else 'partial',local_path=str((folder/'thread.json').relative_to(dest)),assets=assets)
def enrich(row,meta,api,sess):
    dest=old.OUT_ROOT/row['slug'];record=read(dest/'ADDITIONS.json',{})
    if record.get('version')==VERSION and record.get('status')=='ok':return record
    record=dict(version=VERSION,slug=row['slug'],competition_id=int(row['id']),updated_at=now())
    try:
        comp=request(sess,'POST','https://www.kaggle.com/api/i/competitions.CompetitionService/GetCompetition',{'competitionName':row['slug']})
        if int(comp.get('id',0))!=int(row['id']):raise RemoteError('competition id mismatch')
        atomic(dest/'competition-live.json',comp)
    except Exception as e:comp={};record['competition_error']=str(e)
    try:
        pages=request(sess,'GET','https://api.kaggle.com/v1/competitions/'+row['slug']+'/pages')
        atomic(dest/'data'/'pages.json',dict(fetched_at=now(),source_url=row['url']+'/data',**pages))
        for page in pages.get('pages',[]):
            if page.get('content'):atomic(dest/'data'/'pages'/(old.safe_name(page.get('name','page'))+'.md'),markdown(page['content']))
        selected=[p for p in pages.get('pages',[]) if p.get('name')=='data-description']
        if not selected or not selected[0].get('content'):raise RemoteError('data-description absent')
        p=selected[0];atomic(dest/'data'/'overview.md','# '+row['title']+' — Data\n\n来源：'+row['url']+'/data\n\n采集时间：'+now()+'\n\n'+markdown(p['content']))
        assets=page_assets(p['content'],row['url']+'/data',dest/'data'/'assets')
        atomic(dest/'data'/'overview.md','# '+row['title']+' — Data\n\n来源：'+row['url']+'/data\n\n采集时间：'+now()+'\n\n'+local_markdown(p['content'],assets,dest/'data'))
        record['data']=dict(status='ok' if all(a['status'] in ('ok','delivered_external') for a in assets) else 'partial',local_path='data/overview.md',fetched_at=now(),assets=assets)
    except Exception as e:record['data']=dict(status='unavailable',error=str(e),fallback='data/overview-meta.md' if (dest/'data'/'overview-meta.md').exists() else None)
    try:
        raw=request(sess,'POST','https://www.kaggle.com/api/i/competitions.LeaderboardService/GetLeaderboard',{'competitionId':int(row['id'])})
        if not any(k in raw for k in ('publicLeaderboard','privateLeaderboard','teams')) and comp.get('hasLeaderboard',True):raise RemoteError('leaderboard empty envelope')
        atomic(dest/'leaderboard'/'raw.json',dict(fetched_at=now(),**raw))
        lb=normalize_board(raw,int(comp['totalTeams']) if 'totalTeams' in comp else None)
        if comp.get('hasLeaderboard') is False:lb['status']='not_applicable'
        lb.update(fetched_at=now(),source_url=row['url']+'/leaderboard',competition_id=int(row['id']))
        solutions=[]
        for team in raw.get('teams',[]):
            if not (team.get('solutionWriteUpUrl') or team.get('writeUpForumTopicId')):continue
            try:solutions.append(solution_thread(api,sess,row,team,dest))
            except Exception as e:solutions.append(dict(team_id=team.get('teamId'),topic_id=team.get('writeUpForumTopicId'),url=team.get('solutionWriteUpUrl'),status='failed',error=str(e)[:300]))
        lb['solutions']=solutions
        atomic(dest/'leaderboard'/'leaderboard.json',lb)
        record['leaderboard']=dict(status=lb['status'],public_rows=len(lb['publicLeaderboard']),private_rows=len(lb['privateLeaderboard']),expected_teams=lb['expected_teams'],solutions=len(solutions),solution_failures=sum(s['status']!='ok' for s in solutions))
    except Exception as e:record['leaderboard']=dict(status='failed',error=str(e)[:300])
    try:
        kernels,capped=old.list_kernels(api,row['slug'])
        atomic(dest/'code'/'votes.json',dict(source='live Code list',fetched_at=now(),count=len(kernels),complete=not capped,items=[dict(ref=k['ref'],votes=k.get('votes')) for k in kernels]))
        record['code_votes']=dict(status='ok' if not capped and all(k.get('votes') is not None for k in kernels) else 'partial',count=len(kernels),fetched_at=now())
    except Exception as e:record['code_votes']=dict(status='failed',error=str(e)[:300])
    record['status']='ok' if all(record[k]['status'] in ('ok','not_applicable') for k in ('data','leaderboard','code_votes')) and not record['leaderboard'].get('solution_failures') else 'partial'
    atomic(dest/'ADDITIONS.json',record);return record

def bounded_enrich(row,meta,api,sess,progress):
    p=old.OUT_ROOT/row['slug']/'ADDITIONS.json';saved=read(p,{})
    if saved.get('version')==VERSION and saved.get('status')=='ok':return saved
    try:
        with metadata_deadline(progress),metadata_competition_lock(row['slug']):
            return enrich(row,meta,api,sess)
    except MetadataTimeout as e:
        # Preserve prior section results and all partial files. Retry this item
        # on the next pass while allowing the rest of the catalog to proceed.
        saved.update(version=VERSION,slug=row['slug'],competition_id=int(row['id']),status='partial',updated_at=now(),metadata_error=str(e),retry_required=True)
        legacy_write(p,saved);print('METADATA_DEFER',row['slug'],str(e),flush=True)
        return saved

def metadata_worker(rows):
    previous=read(STATE/'metadata-status.json',{});checkpoint=read(STATE/'metadata-checkpoint.json',{})
    start=metadata_resume_index(rows,previous,checkpoint);digest=catalog_digest(rows)
    status('metadata','starting',total=len(rows))
    install_metadata_transport()
    from kaggle.api.kaggle_api_extended import KaggleApi
    api=KaggleApi();api.authenticate();sess=old.session_from_kaggle_json()
    meta=metadata_rows()
    offline_ready=(checkpoint.get('catalog_digest')==digest and checkpoint.get('offline_ready')) or previous.get('phase') in ('live_backfill','pass_finished')
    if not offline_ready:
        db=build_vote_db()
        try:bootstrap(rows,meta,db)
        finally:db.close()
    while True:
        counts=Counter()
        for row in rows[:start]:
            record=read(old.OUT_ROOT/row['slug']/'ADDITIONS.json',{})
            if record.get('status'):counts[record['status']]+=1
        for index in range(start,len(rows)):
            row=rows[index];i=index+1
            wait_room('metadata');status('metadata','live_backfill',done=i-1,total=len(rows),current_slug=row['slug'],counts=dict(counts))
            atomic(STATE/'metadata-checkpoint.json',dict(catalog_digest=digest,next_index=index,current_slug=row['slug'],offline_ready=True,updated_at=now()))
            progress=dict(done=index,total=len(rows),current_slug=row['slug'],counts=dict(counts))
            result=bounded_enrich(row,meta.get(str(row['id']),{}),api,sess,progress);counts[result['status']]+=1
            atomic(STATE/'metadata-checkpoint.json',dict(catalog_digest=digest,next_index=index+1,offline_ready=True,updated_at=now()))
            status('metadata','live_backfill',done=i,total=len(rows),current_slug=row['slug'],counts=dict(counts))
            print('ENRICH',i,len(rows),row['slug'],result['status'],flush=True)
        status('metadata','pass_finished',done=len(rows),total=len(rows),counts=dict(counts))
        if counts['ok']==len(rows):atomic(STATE/'metadata-status.json',dict(state='completed',updated_at=now(),counts=dict(counts)));return
        start=0;atomic(STATE/'metadata-checkpoint.json',dict(catalog_digest=digest,next_index=0,offline_ready=True,updated_at=now()))
        time.sleep(3600)

def assets_worker(rows):
    wait_room('assets',True)
    status('assets','validating_local_cache',total=len(rows))
    from kaggle.api.kaggle_api_extended import KaggleApi
    api=KaggleApi();api.authenticate();sess=old.session_from_kaggle_json();install_safety(api)
    for marker in old.OUT_ROOT.glob('*/models/*/.ok'):
        records=read(marker.parent/'.model-files.json',{})
        for ref,item in records.items():
            p=marker.parent/item['file']
            if p.is_file() and p.stat().st_size==item['bytes'] and gzip_ok(p):MODEL_CACHE.setdefault(ref,p)
        archives=list(marker.parent.glob('*.tar.gz'))
        if len(archives)==1 and gzip_ok(archives[0]):MODEL_CACHE.setdefault(marker.read_text().strip(),archives[0])
    for marker in old.OUT_ROOT.glob('*/code/*/.ok'):
        SOURCE_CACHE.setdefault(marker.read_text().strip(),marker.parent)
    old.ensure_code_index({r['id'] for r in rows},old.STATE_DIR/'code_index.jsonl')
    old.ensure_kernel_versions();old.ensure_forum_db()
    while True:
        counts=Counter()
        for i,row in enumerate(rows,1):
            p=old.OUT_ROOT/row['slug']/'VERIFY-v2.json';prev=read(p,{})
            if prev.get('ok') and prev.get('safety_version')==VERSION:counts['ok']+=1;continue
            wait_room('assets',True);status('assets','collecting',done=i-1,total=len(rows),current_slug=row['slug'],counts=dict(counts))
            try:
                # Only this competition is protected from receipt cleanup while in use.
                handoff_lock=Path('/var/lib/dataset-move/competition-locks')/(row['slug']+'.lock')
                handoff_lock.parent.mkdir(parents=True,exist_ok=True)
                with handoff_lock.open('a') as lease:
                    fcntl.flock(lease,fcntl.LOCK_EX)
                    report=old.download_competition(row,api,sess,force=False)
                report.update(safety_version=VERSION,checked_at=now());atomic(p,report)
                counts['ok' if report.get('ok') else 'partial']+=1
            except old.DiskLimit:
                wait_room('assets',True);break
            except Exception as e:
                atomic(p,dict(slug=row['slug'],ok=False,error=type(e).__name__+': '+str(e)[:300],checked_at=now(),safety_version=VERSION));counts['failed']+=1
        else:
            status('assets','pass_finished',done=len(rows),total=len(rows),counts=dict(counts))
            if counts['ok']==len(rows):atomic(STATE/'assets-status.json',dict(state='completed',updated_at=now(),counts=dict(counts)));return
            time.sleep(3600)

def publish_worker(rows):
    while True:
        additions=[];assets=[]
        for row in rows:
            dest=old.OUT_ROOT/row['slug']
            a=read(dest/'ADDITIONS.json')
            if a:additions.append(a)
            v=read(dest/'VERIFY-v2.json')
            if v:assets.append(dict(slug=row['slug'],ok=v.get('ok'),code=v.get('code'),models=v.get('models'),discussion=v.get('discussion'),error=v.get('error')))
        summary=dict(updated_at=now(),planned_competitions=len(rows),additions_checked=len(additions),additions_status=dict(Counter(a['status'] for a in additions)),assets_checked=len(assets),assets_status=dict(Counter('ok' if a['ok'] else 'partial' for a in assets)),metadata_worker=read(STATE/'metadata-status.json'),assets_worker=read(STATE/'assets-status.json'),scope='all catalog competitions; Code, Models, Discussion, Data-page Markdown, full Public/Private leaderboards and linked Solution writeups/attachments; no official raw competition data bundles')
        summary['confirmed_delivery_cleanup']=read(STATE/'cleanup-summary.json')
        atomic(STATE/'summary.json',summary)
        atomic(STATE/'delivery.json',dict(summary=summary,additions=additions,assets=assets))
        doc='<!doctype html><meta charset="utf-8"><title>Kaggle 全量采集进度</title><style>body{max-width:1100px;margin:30px auto;font:16px/1.6 system-ui}pre{white-space:pre-wrap}</style><h1>Kaggle 全量采集与新增需求回填</h1><p>任务运行完成与资源全部齐全分别统计；未采到不标完整。原有文件保留，空间不足自动等待。</p><p><a href="download_state/full-v2/delivery.json">逐比赛交付清单</a> · <a href="competitions/">比赛目录</a></p><pre>'+html.escape(json.dumps(summary,ensure_ascii=False,indent=2))+'</pre>'
        atomic(ROOT/'full-delivery.html',doc)
        time.sleep(60)

def main():
    ap=argparse.ArgumentParser();ap.add_argument('mode',choices=['metadata','assets','publish']);ap.add_argument('--slug',action='append');args=ap.parse_args()
    STATE.mkdir(parents=True,exist_ok=True)
    lock=(STATE/(args.mode+'.lock')).open('a');fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
    old.MAX_RETRIES=3;old.BACKOFF_CAP=30
    rows=old.load_competitions(old.CSV_PATH)
    if args.slug:rows=[r for r in rows if r['slug'] in args.slug]
    try:{'metadata':metadata_worker,'assets':assets_worker,'publish':publish_worker}[args.mode](rows)
    except BaseException as e:
        atomic(STATE/(args.mode+'-status.json'),dict(state='failed',updated_at=now(),error=type(e).__name__+': '+str(e)[:300]));raise
if __name__=='__main__':main()