"""下载代理共用资源/有效价格/当前所有权核查,不把存储故障视为免费。""" 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)