proxy_access.py

"""下载代理共用资源/有效价格/当前所有权核查，不把存储故障视为免费。"""
from owned_thread import OwnedThreadGroup
from models_sql import Ownership
from purchase_transaction import effective_resource_price


class ProxyAccessError(Exception):
    def __init__(self, status_code, message):
        self.status_code = status_code
        self.message = message
        super().__init__(message)


def read_proxy_item(store, item_id):
    try:
        source = store.load_data('items.json', default_data=[], strict=True, skip_cache=True)
        if not isinstance(source, list):
            raise ValueError('invalid resource source')
        if any(not isinstance(item, dict) or not isinstance(item.get('id'), str) or not item['id'] for item in source):
            raise ValueError('invalid resource record')
        matches = [item for item in source if item.get('id') == item_id]
        if not matches:
            raise ProxyAccessError(404, '资源不存在')
        if len(matches) != 1:
            raise ValueError('ambiguous resource')
        item = matches[0]
        author = item.get('author')
        if not isinstance(author, str) or not author.strip():
            raise ValueError('invalid resource author')
        effective_resource_price(item)
        return item
    except ProxyAccessError:
        raise
    except Exception:
        raise ProxyAccessError(503, '资源信息暂时无法核实，请稍后核对') from None


def _authorized_proxy_item(store, session, item_id, account):
    try:
        item = read_proxy_item(store, item_id)
        author = item['author']
        price = effective_resource_price(item)
        if price > 0 and account != author:
            # SQL会话仍留在调用线程；不用陈旧的身份/拥有缓存，也不授予已退款记录。
            ownerships = session.query(Ownership).filter(Ownership.account == account, Ownership.item_id == item_id).populate_existing().all()
            if any(type(row.is_refunded) is not bool for row in ownerships):
                raise ValueError('invalid ownership refund state')
            rows = [row for row in ownerships if not row.is_refunded]
            if len(rows) > 1:
                raise ValueError('ambiguous ownership')
            if not rows:
                raise ProxyAccessError(403, '尚未获取该资源')
        return item
    except ProxyAccessError:
        raise
    except Exception:
        raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
    finally:
        try:
            session.rollback()  # 不让下载流长期持有所有权读取事务。
        except Exception:
            raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None


MAX_PROXY_ACCESS = 8
_proxy_access_io = OwnedThreadGroup(MAX_PROXY_ACCESS,
    lambda: ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对'))
start_proxy_access = _proxy_access_io.start
stop_proxy_access = _proxy_access_io.stop
close_proxy_access = _proxy_access_io.close


async def authorized_proxy_item(store, session_factory, item_id, account):
    def execute():
        try:
            with session_factory() as session:
                return _authorized_proxy_item(store, session, item_id, account)
        except ProxyAccessError:
            raise
        except Exception:
            raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
    return await _proxy_access_io.run(execute)