import hashlib
import io
import json
import re
import secrets
import sqlite3
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from urllib.parse import urlsplit

import openpyxl
import pytest
from flask import Flask
from sqlalchemy import text
from werkzeug.datastructures import MultiDict

from conftest import PASSWORD, login, post


def order(app, oid):
    with app.Session() as db:
        o = db.get(app.Order, oid)
        return {'status': o.status, 'version': o.row_version, 'revision': o.revision,
                'total': o.total_amount, 'supplier': o.supplier_name,
                'items': [{'qty': i.qty, 'price': i.price, 'date': i.delivery_date} for i in o.items]}


def new(app, client, kind='purchase', key=None):
    response = post(client, '/orders/new', {'doc_type': kind, 'request_key': key or secrets.token_hex(16)})
    assert response.status_code == 302, response.get_data(as_text=True)
    return int(re.search(r'/orders/(\d+)', response.location)[1])


def form(version, qty='3', price='12.5', n=1, **overrides):
    values = MultiDict({'row_version': str(version), 'supplier_name': 'Audit supplier', 'our_company': '苏州博特维智能科技有限公司',
                       'payment_terms': '月结30天', 'delivery_place': '苏州工业园区', 'order_date': '2026-09-16'})
    for i in range(n):
        for k,v in {'code': 'TEST-%03d' % i, 'spec': '测试物料 A&B <规格>', 'unit': '件',
                    'qty': qty, 'price': price, 'date': '2026-10-01', 'remark': ''}.items():
            values.add('item_' + k, v)
    for key, value in overrides.items():
        values[key] = value
    return values


def save(app, client, oid, **kwargs):
    return post(client, '/orders/%s/edit' % oid, form(order(app, oid)['version'], **kwargs))


def action(app, client, oid, name, **kwargs):
    return post(client, '/orders/%s/%s' % (oid, name), {'row_version': str(order(app, oid)['version']), **kwargs})


def pending(app, clients, **kwargs):
    oid = new(app, clients['maker'])
    assert save(app, clients['maker'], oid, intent='submit', **kwargs).status_code == 302
    return oid


def approve(app, clients, oid):
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302


@pytest.mark.parametrize('kind,n', [('purchase', 3), ('outsourcing', 30)])
def test_full_flow_and_frozen_pdf(app, clients, kind, n):
    from models import OrderVersion
    oid = new(app, clients['maker'], kind)
    assert save(app, clients['maker'], oid, n=n, intent='submit').status_code == 302
    assert order(app, oid)['status'] == 'pending_approval'
    assert action(app, clients['maker'], oid, 'approve', stamped='1').status_code == 403
    assert action(app, clients['approver'], oid, 'approve').status_code == 400
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
    assert order(app, oid)['status'] == 'approved'
    response = clients['maker'].get('/orders/%s/pdf' % oid)
    assert response.data.startswith(b'%PDF') and response.data.count(b'%PDF-') == 1
    assert clients['maker'].get('/orders/%s/pdf' % oid).data == response.data
    import pymupdf
    with pymupdf.open(stream=response.data, filetype='pdf') as doc:
        # Older PDF readers may split adjacent Chinese glyphs into lines.
        content = re.sub(r'\s+', '', ''.join(p.get_text() for p in doc))
        assert content.count('采购专用章') == 1
        assert '审核专用章' not in content and '批准专用章' not in content
    with app.Session() as db:
        v = db.query(OrderVersion).filter_by(order_id=oid, revision=1).one()
        assert v.pdf_sha256 == hashlib.sha256(response.data).hexdigest()
        assert len(json.loads(v.content_json)['items']) == n
        assert v.template_version == 'po-v5'
    assert save(app, clients['maker'], oid).status_code == 403
    assert clients['maker'].get('/orders/%s/edit' % oid).status_code == 302
    assert clients['maker'].get('/orders/%s/versions/1' % oid).status_code == 200


def test_cross_maker_access_all_paths(app, clients):
    oid = pending(app, clients)
    number = None
    with app.Session() as db:
        number = db.get(app.Order, oid).order_no
    other = clients['other']
    for suffix in ('view','edit','pdf','versions/1'):
        assert other.get('/orders/%s/%s' % (oid,suffix)).status_code == 404
    assert number not in other.get('/orders').get_data(as_text=True)
    assert 'Audit supplier' not in other.get('/stats').get_data(as_text=True)
    ledger = openpyxl.load_workbook(io.BytesIO(other.get('/export/ledger').data)).active
    assert ledger.max_row == 1
    assert action(app, other, oid, 'copy', request_key=secrets.token_hex(16)).status_code == 404
    assert other.get('/exports/old.pdf').status_code == 404


def test_maker_separation_single_approval_and_rejection(app, clients):
    for name in ('reviewer','approver'):
        assert post(clients[name], '/orders/new', {'request_key': secrets.token_hex(16)}).status_code == 403
    oid = new(app, clients['admin'])
    assert save(app, clients['admin'], oid, intent='submit').status_code == 302
    assert action(app, clients['admin'], oid, 'approve', stamped='1').status_code == 403
    assert action(app, clients['admin'], oid, 'reject', comment='self').status_code == 403
    assert action(app, clients['reviewer'], oid, 'reject', comment='需要修改').status_code == 302
    assert order(app, oid)['status'] == 'rejected'
    assert save(app, clients['admin'], oid, intent='submit').status_code == 302
    assert order(app, oid)['revision'] == 2
    assert action(app, clients['reviewer'], oid, 'approve', stamped='1').status_code == 302
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 403
    assert action(app, clients['approver'], oid, 'reject', comment='already done').status_code == 403
    oid2 = pending(app, clients)
    assert action(app, clients['admin'], oid2, 'approve', stamped='1').status_code == 302


def test_old_review_form_cannot_finally_approve(app, clients):
    oid = pending(app, clients)
    before = order(app, oid)
    assert action(app, clients['reviewer'], oid, 'review', stamped='1').status_code == 409
    assert order(app, oid) == before
    with app.Session() as db:
        assert not db.get(app.Order, oid).approved_by


def test_pdf_failure_rolls_back_approval_and_notification(app, clients, monkeypatch):
    import order_routes
    from models import Notification, OrderVersion
    from test_notifications import configure
    configure(clients)
    oid = pending(app, clients)
    before = order(app, oid)
    def fail(order):
        raise ValueError('PDF generation failed')
    monkeypatch.setattr(order_routes, 'pdf_bytes', fail)
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 400
    assert order(app, oid) == before
    with app.Session() as db:
        assert not db.get(app.Order, oid).approved_by
        assert db.query(Notification).count() == 1
        assert db.query(OrderVersion).filter_by(order_id=oid).one().pdf_data is None


@pytest.mark.parametrize('qty,price', [('-2','10'),('abc','10'),('nan','10'),('Infinity','10'),('1','-1'),('1','0.00001')])
def test_bad_numbers_do_not_replace_saved_order(app, clients, qty, price):
    oid = new(app, clients['maker'])
    assert save(app, clients['maker'], oid).status_code == 302
    before = order(app, oid)
    assert save(app, clients['maker'], oid, qty=qty, price=price).status_code == 400
    assert order(app, oid) == before


@pytest.mark.parametrize('changes', [{'supplier_name':''},{'qty':'0'},{'price':'0'},{'order_date':'bad'}])
def test_submit_validation_atomic(app, clients, changes):
    oid = new(app, clients['maker'])
    assert save(app, clients['maker'], oid).status_code == 302
    before = order(app, oid)
    assert save(app, clients['maker'], oid, intent='submit', **changes).status_code == 400
    assert order(app, oid) == before


def test_money_and_save_submit_latest_values(app, clients):
    oid = new(app, clients['maker'])
    assert save(app, clients['maker'], oid, qty='1',price='0.334',n=2).status_code == 302
    assert order(app, oid)['total'] == Decimal('0.66')
    assert save(app, clients['maker'], oid, qty='3',price='0.3333',intent='submit').status_code == 302
    assert order(app, oid)['total'] == Decimal('1.00')
    assert order(app, oid)['items'][0]['price'] == Decimal('0.3333')
    with app.engine.connect() as con:
        assert con.execute(text('SELECT total_cents, typeof(total_cents) FROM orders WHERE id=:id'), {'id': oid}).one() == (100,'integer')
    from money import line_amount
    assert line_amount('1','2.675') == Decimal('2.68')


def workbook(bad=False):
    wb = openpyxl.Workbook()
    wb.active.append(['物料代码','物料规格','单位','数量','含税单价','交期','备注'])
    wb.active.append(['X','日期测试','件', 'oops' if bad else 3, 0.3333, datetime(2026,10,1),''])
    out = io.BytesIO()
    wb.save(out)
    out.seek(0)
    return out


def test_chinese_filename_excel_date_and_atomic_import(app, clients):
    oid = new(app, clients['maker'])
    response = action(app, clients['maker'], oid, 'import_excel', mode='replace', file=(workbook(), '采购明细.xlsx'))
    assert response.status_code == 302
    assert order(app, oid)['items'] == [{'qty': Decimal('3'), 'price': Decimal('0.3333'), 'date':'2026-10-01'}]
    before = order(app, oid)
    assert action(app, clients['maker'], oid, 'import_excel', mode='replace', file=(workbook(True), '采购明细.xlsx')).status_code == 400
    assert order(app, oid) == before


def test_xls_native_date_and_masterdata_chinese_uploads(app, clients):
    import xlwt
    wb=xlwt.Workbook(); ws=wb.add_sheet('Sheet1')
    for i,h in enumerate(['物料代码','物料规格','单位','数量','单价','交期','备注']): ws.write(0,i,h)
    for i,v in enumerate(['XLS','旧版日期','件',2,3]): ws.write(1,i,v)
    ws.write(1,5,datetime(2026,10,1),xlwt.easyxf(num_format_str='YYYY-MM-DD'))
    out=io.BytesIO(); wb.save(out); out.seek(0)
    oid=new(app,clients['maker'])
    assert action(app,clients['maker'],oid,'import_excel',file=(out,'中文明细.xls')).status_code==302
    assert order(app,oid)['items'][0]['date']=='2026-10-01'
    for endpoint in ['materials','suppliers']:
        template=clients['admin'].get('/%s/import_template.xlsx'%endpoint).data
        response=post(clients['admin'],'/%s/import_excel'%endpoint,{'file':(io.BytesIO(template),'中文模板.xlsx')})
        assert response.status_code==302
    with app.Session() as db:
        assert db.query(app.Material).count()==2
        assert db.query(app.Supplier).count()==1


def test_old_secret_csrf_disabled_and_changed_password_sessions(app, clients):
    shadow = Flask('old')
    shadow.secret_key = 'botw-po-system-2026'
    cookie = shadow.session_interface.get_signing_serializer(shadow).dumps({'user_id':1,'session_version':1})
    forged = app.app.test_client()
    forged.set_cookie('session', cookie)
    assert urlsplit(forged.get('/users').location).path == '/login'
    assert clients['maker'].post('/orders/new', data={'request_key':secrets.token_hex(16)}).status_code == 400
    assert clients['maker'].post('/orders/new', data={'csrf_token':'无效标记'}).status_code == 400
    with app.Session() as db:
        u = db.query(app.User).filter_by(username='reviewer').one()
        uid = u.id
    oid = pending(app, clients)
    assert post(clients['admin'], '/users', {'id':uid,'username':'reviewer','name':'reviewer','role':'approver','active':'0'}).status_code == 302
    assert urlsplit(action(app, clients['reviewer'], oid, 'approve', stamped='1').location).path == '/login'
    assert post(clients['admin'], '/users', {'id':uid,'username':'reviewer','name':'reviewer','role':'approver','active':'1','password':'Reset-Password-2026'}).status_code == 302
    c = app.app.test_client()
    c.get('/login')
    assert urlsplit(post(c,'/login',{'username':'reviewer','password':'Reset-Password-2026'}).location).path == '/account/password'
    c.get('/account/password')
    assert post(c,'/account/password',{'old_password':'Reset-Password-2026','password':PASSWORD,'confirm_password':PASSWORD}).status_code == 302
    c.get('/')
    assert action(app, c, oid, 'approve', stamped='1').status_code == 302


def test_failed_db_write_isolated_and_last_admin_protected(app, clients):
    result = post(clients['admin'], '/users', {'username':'maker','name':'duplicate','role':'maker','password':PASSWORD})
    assert result.status_code == 409
    assert clients['maker'].get('/orders').status_code == 200
    with app.Session() as db:
        uid = db.query(app.User).filter_by(username='admin').one().id
    assert post(clients['admin'], '/users', {'id':uid,'username':'admin','name':'admin','role':'maker','active':'1'}).status_code == 400


def test_withdraw_revise_void_and_preserved_versions(app, clients):
    from models import OrderVersion
    oid = pending(app, clients)
    assert action(app, clients['maker'], oid, 'withdraw', comment='修改价格').status_code == 302
    assert save(app, clients['maker'], oid, price='20', intent='submit').status_code == 302
    approve(app, clients, oid)
    old_pdf = clients['maker'].get('/orders/%s/pdf' % oid).data
    assert action(app, clients['maker'], oid, 'revise', comment='再次变更').status_code == 302
    assert order(app, oid)['status'] == 'draft'
    assert clients['maker'].get('/orders/%s/pdf' % oid).data != old_pdf
    with app.Session() as db:
        v = db.query(OrderVersion).filter_by(order_id=oid, revision=2).one()
        assert v.pdf_data == old_pdf
        assert json.loads(v.content_json)['items'][0]['price'] == '20.0000'
    assert save(app, clients['maker'], oid, price='21', intent='submit').status_code == 302
    approve(app, clients, oid)
    assert order(app, oid)['revision'] == 3
    assert action(app, clients['approver'], oid, 'void', comment='停止采购').status_code == 302
    assert order(app, oid)['status'] == 'voided'
    assert save(app, clients['maker'], oid).status_code == 403
    assert clients['maker'].get('/orders/%s/pdf' % oid).data != old_pdf


def test_stale_forms_and_concurrent_number_allocation(app, clients):
    oid = new(app, clients['maker'])
    old = order(app,oid)['version']
    assert save(app, clients['maker'], oid).status_code == 302
    assert post(clients['maker'], '/orders/%s/edit'%oid, form(old)).status_code == 409
    sessions = [login(app, 'maker') for _ in range(8)]
    with ThreadPoolExecutor(max_workers=8) as pool:
        ids = list(pool.map(lambda c: new(app,c), sessions))
    assert len(set(ids)) == 8
    with app.Session() as db:
        numbers = [o.order_no for o in db.query(app.Order)]
    assert len(set(numbers)) == len(numbers)
    key = secrets.token_hex(16)
    assert new(app, clients['maker'],key=key) == new(app,clients['maker'],key=key)


def test_simultaneous_approval_has_one_winner(app, clients):
    oid = pending(app,clients)
    revision = order(app,oid)['version']
    reviewers = [login(app, name) for name in ('reviewer', 'approver')]
    def approve_once(c):
        return post(c, '/orders/%s/approve'%oid, {'row_version':str(revision),'stamped':'1'}).status_code
    with ThreadPoolExecutor(max_workers=2) as pool:
        statuses = list(pool.map(approve_once, reviewers))
    assert sorted(statuses) == [302,409]
    with app.Session() as db:
        assert db.query(app.ApprovalLog).filter_by(order_id=oid,action='审批通过并归档正式 PDF').count() == 1


def test_pages_and_all_post_forms_have_csrf(app, clients):
    from html.parser import HTMLParser
    class Forms(HTMLParser):
        def __init__(self):
            super().__init__(); self.active=None; self.forms=[]
        def handle_starttag(self,tag,attrs):
            attrs=dict(attrs)
            if tag=='form' and attrs.get('method','').lower()=='post': self.active=set()
            if tag=='input' and self.active is not None: self.active.add(attrs.get('name'))
        def handle_endtag(self,tag):
            if tag=='form' and self.active is not None: self.forms.append(self.active); self.active=None
    oid = new(app,clients['maker'])
    for page in ['/','/orders','/orders/new','/orders/%s/edit'%oid,'/orders/%s/view'%oid,
                 '/users','/users?edit=2','/suppliers','/materials','/stats','/account/password','/settings/notifications']:
        response=clients['admin'].get(page)
        assert response.status_code==200,(page,response.status_code)
        parser=Forms();parser.feed(response.get_data(as_text=True))
        assert all('csrf_token' in fields for fields in parser.forms),page


def test_setup_requires_local_token(app):
    with app.Session() as db:
        db.query(app.User).delete(); db.commit()
    c=app.app.test_client(); c.get('/setup')
    body={'username':'owner','name':'Owner','password':PASSWORD,'setup_token':'wrong'}
    assert post(c,'/setup',body).status_code==403
    body['setup_token']=(Path(app.INSTANCE)/'setup.token').read_text().strip()
    assert post(c,'/setup',body).status_code==302
    assert c.get('/setup').location.endswith('/login')