"""V8.5 restricted MySQL connector. Only approved tables and selected columns.

Database-side SELECT-only account and TLS are mandatory. Results are staged
through V8 import batches; no connector directly publishes or edits content.
"""
from __future__ import annotations
import json
import os
import re
import ssl
import uuid
from fastapi import APIRouter, Request, Body, HTTPException
from . import db, enterprise, jobs, secret_refs, v8_workflows, redis_signal

router=APIRouter(prefix='/api/connectors/sites',tags=['V8.5 read-only MySQL'])
IDENT=re.compile(r'^[A-Za-z_][A-Za-z0-9_]{0,63}$')
HOST=re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9.-]{0,252}$')
ALLOWED_TYPES={'product','vehicle','post','page','brand','case','service','other'}


def setup():
    with db.connect() as c:
        c.executescript('''
        CREATE TABLE IF NOT EXISTS v85_mysql_sources (
          id TEXT PRIMARY KEY,site_id TEXT NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
          host TEXT NOT NULL,port INTEGER NOT NULL,db_name TEXT NOT NULL,
          table_name TEXT NOT NULL,key_column TEXT NOT NULL,columns_json TEXT NOT NULL,
          user_ref TEXT NOT NULL,password_ref TEXT NOT NULL,
          mapping_json TEXT NOT NULL,kind TEXT NOT NULL,page_size INTEGER NOT NULL,
          max_pages INTEGER NOT NULL,next_page INTEGER NOT NULL DEFAULT 1,
          cursor_json TEXT NOT NULL DEFAULT 'null',scanned_pages INTEGER NOT NULL DEFAULT 0,
          staged_count INTEGER NOT NULL DEFAULT 0,state TEXT NOT NULL DEFAULT 'ready',
          created_at TEXT NOT NULL,updated_at TEXT NOT NULL,error TEXT NOT NULL DEFAULT '');
        CREATE INDEX IF NOT EXISTS idx_v85_mysql_site ON v85_mysql_sources(site_id,created_at);
        CREATE TABLE IF NOT EXISTS v85_mysql_pages (
          source_id TEXT NOT NULL REFERENCES v85_mysql_sources(id) ON DELETE CASCADE,
          page INTEGER NOT NULL,batch_id TEXT REFERENCES v8_import_batches(id),
          input_count INTEGER NOT NULL,last_cursor_json TEXT NOT NULL,
          created_at TEXT NOT NULL,PRIMARY KEY(source_id,page));
        ''')


def _approved_env(name):
    return {v.strip().lower() for v in os.environ.get(name,'').split(',') if v.strip()}


def _identifier(value):
    if not isinstance(value,str) or not IDENT.fullmatch(value):
        raise ValueError('数据库/表/列标识符不合法')
    return value


def create_source(site_id,payload):
    v8_workflows._require_site(site_id)
    if not isinstance(payload,dict):raise ValueError('MySQL 配置必须为对象')
    host=str(payload.get('host','')).lower().rstrip('.')
    if not HOST.fullmatch(host) or host not in _approved_env('SITEFORGE_MYSQL_ALLOWED_HOSTS'):
        raise ValueError('MySQL 主机未获管理员许可')
    port=payload.get('port',3306)
    if type(port)!=int or not 1<=port<=65535:raise ValueError('数据库端口不合法')
    db_name=_identifier(payload.get('database'))
    table=_identifier(payload.get('table'))
    approved=_approved_env('SITEFORGE_MYSQL_ALLOWED_TABLES')
    if f'{db_name}.{table}'.lower() not in approved:
        raise ValueError('MySQL 表未被管理员加入白名单')
    key=_identifier(payload.get('key_column'))
    columns=payload.get('columns')
    if not isinstance(columns,list) or not 1<=len(columns)<=60 or any(not isinstance(x,str) for x in columns):
        raise ValueError('MySQL 选择列必须为 1-60 个标识符')
    columns=[_identifier(x) for x in columns]
    if len(columns)!=len(set(columns)) or key not in columns:
        raise ValueError('唯一游标列必须属于选择列，且列不得重复')
    kind=payload.get('content_type','product')
    if kind not in ALLOWED_TYPES:raise ValueError('不支持的内容类型')
    size=payload.get('page_size',100);limit=payload.get('max_pages',20)
    if type(size)!=int or type(limit)!=int or not 1<=size<=250 or not 1<=limit<=50 or size*limit>5000:
        raise ValueError('每次导入最多 5000 条、50 页、单页 250 条')
    mapping=payload.get('mapping') or {}
    if not isinstance(mapping,dict) or len(mapping)>80 or any(not isinstance(k,str) or not isinstance(v,str) or v not in columns for k,v in mapping.items()):
        raise ValueError('字段映射必须引用已经批准的列')
    user_ref=secret_refs.validate_reference(payload.get('user_ref'))
    password_ref=secret_refs.validate_reference(payload.get('password_ref'))
    source_id=uuid.uuid4().hex
    db.insert('''INSERT INTO v85_mysql_sources(id,site_id,host,port,db_name,table_name,key_column,
      columns_json,user_ref,password_ref,mapping_json,kind,page_size,max_pages,created_at,updated_at)
      VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''',
      (source_id,site_id,host,port,db_name,table,key,json.dumps(columns),user_ref,password_ref,
       json.dumps(mapping,ensure_ascii=False),kind,size,limit,db.now(),db.now()))
    return status(site_id,source_id)


def _one(site_id,source_id):
    row=db.one('SELECT * FROM v85_mysql_sources WHERE id=? AND site_id=?',(source_id,site_id))
    if row is None:raise HTTPException(404,'MySQL 数据源不存在')
    return row


def status(site_id,source_id):
    row=_one(site_id,source_id)
    # Neither secret reference nor credential is emitted to browsers or logs.
    safe={k:row[k] for k in ('id','site_id','host','db_name','table_name','kind','state',
          'next_page','scanned_pages','staged_count','max_pages','page_size','error')}
    safe['pages']=db.rows('SELECT page,batch_id,input_count FROM v85_mysql_pages WHERE source_id=? ORDER BY page',(source_id,))
    return safe


def relationship_preview(site_id,batch_id):
    """Read-only assessment of brand/category references before batch commit."""
    row=db.one('SELECT id,state FROM v8_import_batches WHERE id=? AND site_id=?',(batch_id,site_id))
    if not row:raise HTTPException(404,'导入批次不存在')
    data=db.rows('SELECT route,content_json FROM v8_import_items WHERE batch_id=? ORDER BY row_number LIMIT 1001',(batch_id,))
    if len(data)>1000:raise ValueError('批次超过预览上限')
    existing_brands={r['title'].strip().casefold() for r in db.rows("SELECT title FROM entries WHERE site_id=? AND type='brand'",(site_id,))}
    batch_brands=set();brands={};categories={};routes={};missing=[]
    for record in data:
        item=json.loads(record['content_json'])
        if item.get('type')=='brand':batch_brands.add(str(item.get('title','')).strip().casefold())
    for record in data:
        item=json.loads(record['content_json']);meta=item.get('meta') or {};fields=meta.get('fields') or {}
        brand=str(fields.get('brand') or '').strip();category=str(fields.get('category') or '').strip()
        if brand:brands[brand]=brands.get(brand,0)+1
        if category:categories[category]=categories.get(category,0)+1
        route=item.get('route','');routes[route]=routes.get(route,0)+1
        if brand and brand.casefold() not in existing_brands|batch_brands:
            missing.append({'route':route,'brand':brand,'reason':'未找到已存在或本批次品牌页面'})
    return {'batch_id':batch_id,'state':row['state'],'items_checked':len(data),
            'brands':brands,'categories':categories,'unlinked_brand_references':missing[:100],
            'unlinked_brand_total':len(missing),
            'duplicate_routes':[k for k,v in routes.items() if v>1],
            'note':'关系预览只提示需要核对的品牌页面，不自动创建或修改记录。'}


def query_for(source,last_cursor):
    """Identifier allowlisting occurs at source creation; no arbitrary SQL input."""
    key=_identifier(source['key_column'])
    cols=json.loads(source['columns_json'])
    col_sql=', '.join('`'+_identifier(x)+'`' for x in cols)
    table='`'+_identifier(source['db_name'])+'`.`'+_identifier(source['table_name'])+'`'
    where=(' WHERE `'+key+'` > %s') if last_cursor is not None else ''
    sql='SELECT '+col_sql+' FROM '+table+where+' ORDER BY `'+key+'` ASC LIMIT %s'
    return sql,((last_cursor,int(source['page_size'])) if last_cursor is not None else (int(source['page_size']),))


def _fetch_page(source,last_cursor):
    # Approved host/table must remain approved when the Worker executes a job.
    if source['host'] not in _approved_env('SITEFORGE_MYSQL_ALLOWED_HOSTS') or (
       source['db_name']+'.'+source['table_name']).lower() not in _approved_env('SITEFORGE_MYSQL_ALLOWED_TABLES'):
        raise ValueError('MySQL 主机或表已经失去授权')
    try:
        import pymysql
    except ImportError as exc:
        raise RuntimeError('MySQL 驱动缺失，请安装 requirements.lock') from exc
    username=secret_refs.resolve(source['user_ref'])
    password=secret_refs.resolve(source['password_ref'])
    # Always verify TLS server certificate and hostname, including Docker use.
    ca=os.environ.get('SITEFORGE_MYSQL_CA_FILE','')
    ctx=ssl.create_default_context(cafile=ca or None)
    ctx.verify_mode=ssl.CERT_REQUIRED
    ctx.check_hostname=True
    sql,params=query_for(source,last_cursor)
    conn=None
    try:
        conn=pymysql.connect(host=source['host'],port=int(source['port']),user=username,
             password=password,database=source['db_name'],ssl=ctx,ssl_verify_cert=True,
             ssl_verify_identity=True,connect_timeout=8,read_timeout=20,write_timeout=8,
             autocommit=True,cursorclass=pymysql.cursors.DictCursor)
        with conn.cursor() as cur:
            cur.execute('START TRANSACTION READ ONLY')
            cur.execute(sql,params)
            rows=cur.fetchall()
        return rows
    except Exception as exc:
        # Never surface driver exception text because it can contain credentials.
        raise ConnectionError('MySQL 安全连接或只读查询失败，请检查 Worker 服务端日志') from exc
    finally:
        if conn is not None:
            try:conn.rollback()
            finally:conn.close()


def _safe_data(rows,columns):
    if not isinstance(rows,(list,tuple)) or len(rows)>250:raise ValueError('MySQL 每页记录超限')
    cleaned=[]
    for item in rows:
        if not isinstance(item,dict):raise ValueError('MySQL 应返回字典格式记录')
        record={}
        for col in columns:
            value=item.get(col)
            if isinstance(value,(dict,list,tuple)):value=json.dumps(value,ensure_ascii=False,default=str)
            if isinstance(value,bytes):value=value.decode('utf-8',errors='replace')
            if value is not None and len(str(value))>20000:raise ValueError('字段超出 20000 字符限制')
            record[col]=None if value is None else str(value)
        cleaned.append(record)
    return cleaned


def scan_page(site_id,source_id):
    src=_one(site_id,source_id)
    if src['state'] in ('cancelled','completed','limit_reached'):
        return {'state':src['state'],'next_page':src['next_page']}
    page=int(src['next_page'])
    if page>src['max_pages']:
        db.execute("UPDATE v85_mysql_sources SET state='limit_reached' WHERE id=?",(source_id,))
        return {'state':'limit_reached'}
    existing=db.one('SELECT * FROM v85_mysql_pages WHERE source_id=? AND page=?',(source_id,page))
    origin=f'mysql:{source_id}:page:{page}'
    if existing:
        count=existing['input_count'];end_cursor=json.loads(existing['last_cursor_json']);batch_id=existing['batch_id']
    else:
        prior=db.one('SELECT id,input_count FROM v8_import_batches WHERE site_id=? AND origin=?',(site_id,origin))
        if prior:
            # Only records staged for this exact source+page can be recovered.
            batch_id=prior['id'];count=prior['input_count']
            last=db.one('''SELECT source_key FROM v8_import_items WHERE batch_id=? ORDER BY row_number DESC LIMIT 1''',(batch_id,))
            if not last:raise ValueError('导入检查点缺少最后来源主键')
            end_cursor=json.loads(last['source_key'])
        else:
            rows=_fetch_page(src,json.loads(src['cursor_json']))
            cols=json.loads(src['columns_json'])
            clean=_safe_data(rows,cols)
            count=len(clean);batch_id=None;end_cursor=json.loads(src['cursor_json'])
            if count:
                keys=[record.get(src['key_column']) for record in clean]
                if any(x is None or x=='' for x in keys) or len(set(keys))!=len(keys):
                    raise ValueError('MySQL 游标列存在空值或重复值；请确保数据库具有唯一索引')
                if end_cursor is not None and str(keys[0])<=str(end_cursor):
                    # Database ORDER BY uses its native type. Numeric keys may be
                    # converted to strings, so compare only for exact equality.
                    if str(keys[0])==str(end_cursor):raise ValueError('游标未向前推进')
                end_cursor=keys[-1]
                mapping=json.loads(src['mapping_json']) or None
                staged=v8_workflows.stage(site_id,clean,mapping,src['kind'],origin,
                                           [json.dumps(k,ensure_ascii=False) for k in keys])
                batch_id=staged['batch_id']
        with db.connect() as c:
            c.execute('BEGIN IMMEDIATE')
            c.execute('''INSERT INTO v85_mysql_pages(source_id,page,batch_id,input_count,last_cursor_json,created_at)
                       VALUES(?,?,?,?,?,?) ON CONFLICT(source_id,page) DO NOTHING''',
                      (source_id,page,batch_id,count,json.dumps(end_cursor,ensure_ascii=False),db.now()))
    final='completed' if count<src['page_size'] else ('limit_reached' if page>=src['max_pages'] else 'ready')
    with db.connect() as c:
        c.execute('BEGIN IMMEDIATE')
        current=c.execute('SELECT state,next_page FROM v85_mysql_sources WHERE id=? AND site_id=?',(source_id,site_id)).fetchone()
        if current and current['state']=='cancelled':return {'state':'cancelled','page':page}
        if current and current['next_page']==page:
            c.execute('''UPDATE v85_mysql_sources SET state=?,next_page=?,cursor_json=?,
                        scanned_pages=scanned_pages+1,staged_count=staged_count+?,updated_at=? WHERE id=?''',
                      (final,page+1,json.dumps(end_cursor,ensure_ascii=False),count,db.now(),source_id))
    return {'page':page,'state':final,'count':count,'next_page':page+1,'batch_id':batch_id}


def run_scan(site_id,source_id,job_id=None):
    while True:
        item=scan_page(site_id,source_id)
        if item['state'] in ('completed','limit_reached','cancelled'):break
    return status(site_id,source_id)


def submit_scan(site_id,source_id,actor,key=None):
    src=_one(site_id,source_id)
    if src['state']=='cancelled':raise ValueError('数据源已取消')
    if key is not None and not (8<=len(key)<=128 and all(c.isalnum() or c in '_-:.' for c in key)):
        raise ValueError('幂等键不合法')
    payload=json.dumps({'source_id':source_id},sort_keys=True,separators=(',',':'))
    with db.connect() as c:
        c.execute('BEGIN IMMEDIATE')
        if key:
            prior=c.execute('SELECT id,state,kind,payload_json FROM build_jobs WHERE site_id=? AND idempotency_key=?',(site_id,key)).fetchone()
            if prior:
                if prior['kind']!='mysql_scan' or prior['payload_json']!=payload:raise ValueError('幂等键已用于其他任务')
                return {'id':prior['id'],'state':prior['state'],'existing':True}
        prior=c.execute("SELECT id,state FROM build_jobs WHERE site_id=? AND kind='mysql_scan' AND payload_json=? AND state IN ('queued','running','retrying')",(site_id,payload)).fetchone()
        if prior:return {'id':prior['id'],'state':prior['state'],'existing':True}
        if src['state'] in ('completed','limit_reached'):
            return {'state':src['state'],'existing':True,'source_id':source_id}
        active="state IN ('queued','running','retrying')"
        if c.execute(f'SELECT COUNT(*) FROM build_jobs WHERE {active}').fetchone()[0]>=jobs._limit('SITEFORGE_MAX_PENDING_JOBS',100):
            raise jobs.JobCapacityError('全局任务容量已满')
        if c.execute(f'SELECT COUNT(*) FROM build_jobs WHERE site_id=? AND {active}',(site_id,)).fetchone()[0]>=jobs._limit('SITEFORGE_MAX_PENDING_JOBS_PER_SITE',5):
            raise jobs.JobCapacityError('当前站点任务容量已满')
        jid=uuid.uuid4().hex[:16]
        c.execute('''INSERT INTO build_jobs(id,site_id,kind,state,created_by,created_at,payload_json,
          idempotency_key,max_attempts,timeout_seconds,available_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)''',
          (jid,site_id,'mysql_scan','queued',actor,db.now(),payload,key,3,900,db.now()))
    redis_signal.notify()
    return {'id':jid,'state':'queued','existing':False}


@router.post('/{site_id}/mysql-sources',status_code=201)
def api_create(request:Request,site_id:str,payload:dict=Body(...)):
    enterprise.require(request,site_id,'owner')
    try:return create_source(site_id,payload)
    except ValueError as exc:raise HTTPException(400,str(exc))

@router.get('/{site_id}/mysql-sources')
def api_list(request:Request,site_id:str):
    enterprise.require(request,site_id,'viewer')
    v8_workflows._require_site(site_id)
    return {'items':db.rows('SELECT id,host,db_name,table_name,state,staged_count FROM v85_mysql_sources WHERE site_id=? ORDER BY created_at DESC LIMIT 100',(site_id,))}

@router.get('/{site_id}/mysql-sources/{source_id}')
def api_status(request:Request,site_id:str,source_id:str):
    enterprise.require(request,site_id,'viewer')
    return status(site_id,source_id)

@router.get('/{site_id}/import-batches/{batch_id}/relationships')
def api_relationships(request:Request,site_id:str,batch_id:str):
    enterprise.require(request,site_id,'viewer')
    return relationship_preview(site_id,batch_id)

@router.post('/{site_id}/mysql-sources/{source_id}/scan',status_code=202)
def api_scan(request:Request,site_id:str,source_id:str,payload:dict=Body(default_factory=dict)):
    enterprise.require(request,site_id,'editor')
    try:return submit_scan(site_id,source_id,request.state.actor,payload.get('idempotency_key'))
    except (ValueError,jobs.JobCapacityError) as exc:raise HTTPException(409,str(exc))

@router.post('/{site_id}/mysql-sources/{source_id}/cancel')
def api_cancel(request:Request,site_id:str,source_id:str):
    enterprise.require(request,site_id,'editor')
    src=_one(site_id,source_id)
    if src['state']!='completed':
        db.execute("UPDATE v85_mysql_sources SET state='cancelled',updated_at=? WHERE id=?",(db.now(),source_id))
    for j in db.rows("SELECT id,payload_json FROM build_jobs WHERE site_id=? AND kind='mysql_scan' AND state IN ('queued','running','retrying')",(site_id,)):
        if json.loads(j['payload_json']).get('source_id')==source_id:jobs.cancel(site_id,j['id'])
    return status(site_id,source_id)