"""Persistent serial queue. Browser submissions are always initiated by the user."""
import json
import copy
import re
from contextlib import contextmanager
from pathlib import Path
import secrets
import sqlite3
import threading
import time
from urllib.parse import urlsplit, parse_qs

from articles import inspect_article

ACCEPTED = {'published', 'accepted_pending_moderation'}
SUBMITTED = {'user_submitting', 'submit_unknown'}


class Store:
    def __init__(self, path, ledger=None):
        self.path = str(path)
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        self.lock = threading.RLock()
        with self.db() as db:
            db.executescript('''
                CREATE TABLE IF NOT EXISTS jobs(id TEXT PRIMARY KEY, article_key TEXT UNIQUE, value TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS ledger(key TEXT PRIMARY KEY, value TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS settings(id INTEGER PRIMARY KEY CHECK(id=1), value TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS posts(id TEXT PRIMARY KEY, job TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS post_aliases(url TEXT PRIMARY KEY, thread_id TEXT NOT NULL, job TEXT NOT NULL);
            ''')
            row = db.execute('SELECT value FROM settings WHERE id=1').fetchone()
            self.run = json.loads(row[0]) if row else dict(running=False, epoch=0, config={}, batch=None, active=None, owner=None, switch=None, reason='请配置任务后开始')
            self.run.update(running=False, epoch=self.run['epoch'] + 1, reason='服务已启动，请明确点击开始/继续；未决任务不会重发')
            for row in db.execute('SELECT value FROM jobs').fetchall():
                job = json.loads(row[0])
                if job['state'] in ACCEPTED and job.get('post_url'):
                    canonical = job.get('thread_id')
                    if not canonical:
                        job['legacy_url_id'] = job.get('post_id')
                        job['identity_needs_reconciliation'] = True
                        self._save(db, job)
                    db.execute('INSERT OR IGNORE INTO post_aliases VALUES(?,?,?)',
                               (job['post_url'], canonical or 'unknown', job['id']))
            self._persist(db)
        self.clients = {}
        if ledger and Path(ledger).is_file():
            self.import_history(json.loads(Path(ledger).read_text()).get('records', []))

    @contextmanager
    def db(self):
        connection = sqlite3.connect(self.path, timeout=20)
        previous = copy.deepcopy(getattr(self, 'run', None))
        try:
            with connection:
                yield connection
        except Exception:
            if previous is not None:
                self.run = previous
            raise
        finally:
            connection.close()

    def _persist(self, db):
        db.execute('INSERT OR REPLACE INTO settings VALUES(1,?)', (json.dumps(self.run, ensure_ascii=False),))

    def _job(self, db, id):
        row = db.execute('SELECT value FROM jobs WHERE id=?', (id,)).fetchone()
        if not row:
            raise ValueError('任务不存在')
        return json.loads(row[0])

    def _save(self, db, job):
        job['updated_at'] = time.time()
        db.execute('INSERT OR REPLACE INTO jobs VALUES(?,?,?)', (job['id'], job['article_key'], json.dumps(job, ensure_ascii=False)))

    def import_history(self, records):
        with self.lock, self.db() as db:
            for record in records:
                digest = record.get('markdown_sha256')
                if digest and record.get('block_repost'):
                    db.execute('INSERT OR IGNORE INTO ledger VALUES(?,?)', ('md:' + digest, json.dumps(record, ensure_ascii=False)))
                    # Add content identity where the original historical file is still present.
                    try:
                        article = inspect_article(Path(record['markdown_path']).parent)
                        db.execute('INSERT OR IGNORE INTO ledger VALUES(?,?)', ('article:' + article['article_key'], json.dumps(record, ensure_ascii=False)))
                    except (OSError, ValueError, KeyError):
                        pass

    def enqueue(self, directory):
        article = inspect_article(directory)
        article.pop('package')
        keys = ['article:' + article['article_key'], 'md:' + article['markdown_sha256']] + ['source:' + str(k) for k in article['source_keys']]
        with self.lock, self.db() as db:
            if any(db.execute('SELECT 1 FROM ledger WHERE key=?', (k,)).fetchone() for k in keys):
                return '已受理，跳过'
            if db.execute('SELECT 1 FROM jobs WHERE article_key=?', (article['article_key'],)).fetchone():
                return '已在队列，跳过'
            # Source URL identities prevent different renderings of one downloaded article re-entering.
            for row in db.execute('SELECT value FROM jobs'):
                if set(article['source_keys']) & set(json.loads(row[0]).get('source_keys', [])):
                    return '相同来源已在队列，跳过'
            article.update(id=secrets.token_hex(12), state='queued', created_at=time.time(), error='', post_url='')
            self._save(db, article)
            return '已入队'

    def state(self):
        with self.lock, self.db() as db:
            jobs = [json.loads(r[0]) for r in db.execute('SELECT value FROM jobs ORDER BY rowid')]
            return dict(run=json.loads(json.dumps(self.run)), jobs=jobs,
                        clients=[v for v in self.clients.values() if time.time() - v['seen'] < 20],
                        history_count=db.execute('SELECT count(*) FROM ledger').fetchone()[0])

    def configure(self, config):
        fid, minimum, maximum = config.get('fid'), config.get('minimum'), config.get('maximum')
        interval = config.get('interval')
        if any(type(x) is not int for x in (fid, minimum, maximum, interval)) or not (fid > 0 and 1 <= minimum <= maximum <= 1000 and 0 <= interval <= 86400):
            raise ValueError('请填写有效版块 ID、篇数范围（1～1000）和间隔秒数')
        if config.get('history_checked') is not True:
            raise ValueError('请先核对待处理文章与历史发布记录，并勾选确认')
        with self.lock, self.db() as db:
            if self.run['active'] or self.run['switch'] or self.run['running']:
                raise ValueError('当前流程未结束，不能更换配置')
            self.run.update(config={k: config[k] for k in ('fid', 'minimum', 'maximum', 'interval', 'history_checked')}, batch=None)
            self._persist(db)

    def control(self, action):
        with self.lock, self.db() as db:
            if action == 'start':
                if not self.run['config']:
                    raise ValueError('请先保存配置')
                self.run.update(running=True, reason='等待浏览器连接')
            else:
                self.run.update(running=False, reason='已终止' if action == 'stop' else '已暂停')
            self.run['epoch'] += 1
            self._persist(db)
            return dict(self.run)

    def _batch(self, account):
        cfg = self.run['config']
        return dict(account=account, target=cfg['minimum'] + secrets.randbelow(cfg['maximum'] - cfg['minimum'] + 1), count=0)

    def _check_owner(self, owner):
        if self.run['owner'] != owner:
            raise ValueError('任务由另一个浏览器标签持有')

    def heartbeat(self, data):
        owner = str(data.get('owner', ''))[:100]
        if not owner:
            raise ValueError('缺少执行标签标识')
        with self.lock:
            self.clients[owner] = dict(owner=owner, account=data.get('account'), url=str(data.get('url', ''))[:1000],
                                       extension=data.get('extension'), seen=time.time())
            return self.state()['run']

    def next(self, data):
        owner, account = data.get('owner'), data.get('account')
        with self.lock, self.db() as db:
            if not self.run['running']:
                return dict(action='idle', run=self.run)
            if not account or not str(account.get('uid', '')).isdigit() or not account.get('name'):
                return dict(action='login', run=self.run)
            if self.run['owner'] not in (None, owner):
                return dict(action='other_tab', run=self.run)
            self.run['owner'] = owner
            if self.run['switch']:
                switch = self.run['switch']
                if account['uid'] != switch['from'] and data.get('extension', {}).get('idle') is True:
                    self.run.update(batch=self._batch(account), switch=None)
                    self._persist(db)
                else:
                    ext = data.get('extension', {})
                    elapsed = time.time() - switch.get('clicked_at', switch['created_at'])
                    if ext.get('paused') or (switch['state'] == 'clicked' and elapsed > 10 and ext.get('idle')) or elapsed > 120:
                        self.run.update(running=False, reason='换号已暂停、超时或账号未改变，请检查插件并人工恢复，禁止重复抽号')
                        self._persist(db)
                        return dict(action='idle', run=self.run)
                    return dict(action='switch' if switch['state'] == 'pending' else 'switch_wait', run=self.run)
            if self.run['batch'] is None:
                self.run['batch'] = self._batch(account)
            if self.run['batch']['account']['uid'] != account['uid']:
                self.run.update(running=False, reason='检测到账号变化，请先核对当前任务')
                self._persist(db)
                return dict(action='idle', run=self.run)
            if self.run.get('return_home'):
                url = urlsplit(str(data.get('url', '')))
                if url.scheme == 'https' and url.netloc in ('www.dww11.com', 'dww11.com') and url.path == '/' and not url.query:
                    self.run['return_home'] = False
                    self._persist(db)
                else:
                    return dict(action='home', run=self.run)
            if self.run['active']:
                job = self._job(db, self.run['active'])
                return dict(action='resume', job=job, run=self.run)
            rows = [json.loads(r[0]) for r in db.execute('SELECT value FROM jobs ORDER BY rowid')]
            job = next((j for j in rows if j['state'] == 'queued'), None)
            if not job:
                self.run.update(running=False, reason='当前队列已处理完；新文章入队后请点击继续')
                self._persist(db)
                return dict(action='idle', run=self.run)
            if time.time() < self.run.get('not_before', 0):
                return dict(action='interval', run=self.run)
            if self.run['batch']['count'] >= self.run['batch']['target']:
                self.run['switch'] = dict(state='pending', **{'from': account['uid']}, created_at=time.time())
                self._persist(db)
                return dict(action='switch', run=self.run)
            job.update(state='preparing', account=account, fid=self.run['config']['fid'], error='')
            self.run['active'] = job['id']
            self._save(db, job)
            self._persist(db)
            return dict(action='prepare', job=job, run=self.run)

    def current(self, data):
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            return {'job': self._job(db, self.run['active']) if self.run['active'] else None, 'run': self.run}

    def skip(self, data):
        """Explicitly discard an unsubmitted task and continue the existing queue."""
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            if data.get('epoch') != self.run['epoch']:
                raise ValueError('流程状态已改变，请重新确认后跳过')
            if self.run['active'] != data.get('id'):
                raise ValueError('不是当前任务，不能跳过')
            job = self._job(db, data['id'])
            if job['state'] not in {'preparing', 'importing', 'formatting', 'awaiting_user_review', 'needs_attention'} or job.get('submitted_at'):
                raise ValueError('文章已提交或结果待核对，不能直接跳过，请先人工核对')
            job.update(state='skipped', error='', skipped_at=time.time())
            self.run.update(active=None, running=True, epoch=self.run['epoch'] + 1,
                            reason='已跳过当前文章，等待准备下一篇')
            self._save(db, job)
            self._persist(db)
            return {'ok': True, 'fid': job['fid'], 'run': dict(self.run)}

    def switch_begin(self, data):
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            if not self.run['running'] or not self.run['switch'] or self.run['switch']['state'] != 'pending':
                raise ValueError('换号状态已改变，禁止重复点击')
            if data.get('epoch', self.run['epoch']) != self.run['epoch']:
                raise ValueError('换号请求已过期')
            self.run['switch'].update(state='clicked', clicked_at=time.time())
            self._persist(db)
            return {'ok': True}

    def package(self, data):
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            if not self.run['running'] or self.run['active'] != data['id']:
                raise ValueError('流程已停止或任务不匹配')
            job = self._job(db, data['id'])
            if job['state'] != 'preparing':
                raise ValueError('当前任务不允许重新导入，请先核对已上传内容')
        article = inspect_article(job['directory'])
        if article['markdown_sha256'] != job['markdown_sha256'] or article['article_key'] != job['article_key']:
            raise ValueError('入队后原文章已改变，请移出并重新扫描')
        return article['package']

    def event(self, data):
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            job = self._job(db, data['id'])
            if self.run['active'] != job['id']:
                raise ValueError('不是当前任务')
            target = data['state']
            transitions = {'preparing': {'importing', 'needs_attention'}, 'importing': {'formatting', 'needs_attention'},
                           'formatting': {'awaiting_user_review', 'needs_attention'},
                           'awaiting_user_review': {'user_submitting', 'needs_attention'},
                           'user_submitting': {'submit_unknown', 'needs_attention', 'rejected'}, 'submit_unknown': {'needs_attention'}}
            if target not in transitions.get(job['state'], set()):
                raise ValueError('任务状态不允许此操作：' + job['state'] + ' → ' + target)
            if target in ('importing', 'formatting', 'awaiting_user_review') and (not self.run['running'] or data.get('epoch') != self.run['epoch']):
                raise ValueError('流程已暂停/终止，过期回调不能继续')
            job.update(state=target, error=str(data.get('error', ''))[:1000])
            if target == 'user_submitting':
                job.update(submitted_at=time.time(), snapshot=data.get('snapshot', {}), attempt=secrets.token_hex(8))
            if target == 'awaiting_user_review':
                job['prepared'] = data.get('snapshot', {})
            if target in ('needs_attention', 'rejected'):
                self.run.update(running=False, reason=job['error'] or '当前任务需要处理')
                self._persist(db)
            self._save(db, job)
            return {'ok': True}

    def reconcile(self, data):
        if data.get('confirmed') is not True:
            raise ValueError('请先确认这是当前文章的已受理结果')
        with self.lock, self.db() as db:
            if self.run['running']:
                raise ValueError('请先暂停流程再补核历史受理记录')
            job = self._job(db, data['id'])
            if data.get('article_key') != job['article_key']:
                raise ValueError('文章身份不匹配')
            return self._accept({**data, 'owner': self.run['owner']}, count_batch=False)

    def accept(self, data):
        return self._accept(data)

    def _accept(self, data, count_batch=True):
        with self.lock, self.db() as db:
            self._check_owner(data['owner'])
            job = self._job(db, data['id'])
            if job['state'] in ACCEPTED:
                return {'ok': True, 'duplicate': True}
            if self.run['active'] != job['id'] or job['state'] not in SUBMITTED:
                raise ValueError('只有已观察到人工提交的任务才能登记结果')
            url = urlsplit(str(data.get('post_url', '')))
            match = re.fullmatch(r'/thread/(\d+)\.html', url.path)
            query = parse_qs(url.query)
            thread_id = str(data.get('thread_id', ''))
            if not re.fullmatch(r'[1-9]\d*', thread_id):
                raise ValueError('缺少详情页真实主题 ID，不能以地址数字代替')
            if url.scheme != 'https' or url.netloc not in ('www.dww11.com', 'dww11.com') or not (match or (url.path == '/forum.php' and query.get('mod') == ['viewthread'] and query.get('tid') == [thread_id])):
                raise ValueError('未识别到帖子详情页')
            if data.get('author_uid') != job['account']['uid'] or data.get('fid') != job['fid']:
                raise ValueError('结果页作者或版块与任务不符，请人工核对')
            if ' '.join(str(data.get('title', '')).split()) != ' '.join(str(job.get('snapshot', {}).get('title', '')).split()):
                raise ValueError('结果页标题与人工提交快照不一致')
            if data.get('state') not in ACCEPTED:
                raise ValueError('需要明确的公开或审核中状态')
            if not data.get('evidence'):
                raise ValueError('缺少页面结果证据')
            evidence = data['evidence']
            if not isinstance(evidence, dict) or str(evidence.get('thread_id')) != thread_id:
                raise ValueError('主题 ID 与页面证据不一致')
            alias = db.execute('SELECT thread_id, job FROM post_aliases WHERE url=?', (url.geturl(),)).fetchone()
            if alias and (alias[0] != thread_id or alias[1] != job['id']):
                raise ValueError('帖子链接已绑定其他记录')
            post_id = thread_id
            existing = db.execute('SELECT job FROM posts WHERE id=?', (post_id,)).fetchone()
            if existing and existing[0] != job['id']:
                raise ValueError('帖子已绑定其他任务，不能重复计数')
            db.execute('INSERT OR IGNORE INTO posts VALUES(?,?)', (post_id, job['id']))
            db.execute('INSERT OR IGNORE INTO post_aliases VALUES(?,?,?)', (url.geturl(), thread_id, job['id']))
            job.update(state=data['state'], thread_id=thread_id, post_id=post_id, first_post_pid=data.get('first_post_pid'), post_url=url.geturl(), result=evidence, integrity_issues=evidence.get('integrity_issues', []))
            keys = ['article:' + job['article_key'], 'md:' + job['markdown_sha256']] + ['source:' + str(k) for k in job['source_keys']]
            for key in keys:
                db.execute('INSERT OR REPLACE INTO ledger VALUES(?,?)', (key, json.dumps(job, ensure_ascii=False)))
            if count_batch:
                self.run['batch']['count'] += 1
            job['historical_reconciliation'] = not count_batch
            self.run.update(active=None, return_home=True, not_before=time.time() + self.run['config']['interval'])
            if job['integrity_issues']:
                self.run.update(running=False, reason='网站已受理，内容核对发现差异，请检查帖子后明确继续')
            self._save(db, job)
            self._persist(db)
            return {'ok': True, 'run': dict(self.run)}

    def resolve(self, data):
        """Explicit local UI resolution, never an automatic retry of an unknown submission."""
        with self.lock, self.db() as db:
            job = self._job(db, data['id'])
            if job['state'] in ACCEPTED:
                raise ValueError('已受理文章不能重置')
            if self.run['running']:
                raise ValueError('请先暂停流程')
            if data.get('confirmed_not_accepted') is not True:
                raise ValueError('必须先人工核对网站未受理，并清理编辑器/草稿')
            job.update(state='queued' if data.get('retry') else 'cancelled', error='', snapshot={})
            for key in ('submitted_at', 'attempt', 'prepared'):
                job.pop(key, None)
            if self.run['active'] == job['id']:
                self.run.update(active=None, owner=None)
            self._save(db, job)
            self._persist(db)
            return {'ok': True}

    def takeover(self, data):
        with self.lock, self.db() as db:
            if self.run['running']:
                raise ValueError('请先在控制台暂停流程，再接管')
            self.run['owner'] = data['owner']
            self._persist(db)
            return {'ok': True}