검색·내보내기·페이지를 같은 자료에 묶는 합성 조회 모듈

"""Independent synthetic read snapshot. No database/auth service or signed cursors.
The company query must be chosen from trusted server context, never client authority.
"""
from dataclasses import dataclass
import json,hashlib,base64,re
def ident(v):return isinstance(v,str) and re.fullmatch('[A-Za-z0-9_-]{1,40}',v) is not None
def packed(v):return json.dumps(v,sort_keys=True,separators=(',',':'),ensure_ascii=False,allow_nan=False)
@dataclass(frozen=True)
class Snapshot:
 company_id:str
 status:str
 query:str
 revision:int
 fingerprint:str
 rows:tuple

def make_snapshot(records,query,revision):
 if not isinstance(records,list) or len(records)>1000 or not isinstance(query,dict) or set(query)-{'company_id','status','q'} or not ident(query.get('company_id')) or type(revision) is not int or not 0<=revision<=9007199254740991:raise ValueError('Invalid snapshot input')
 status=query.get('status','all');q=query.get('q','')
 if status not in ['all','draft','sent'] or not isinstance(q,str) or len(q)>40:raise ValueError('Invalid query')
 q=q.casefold();seen=set();selected=[]
 for row in records:
  if not isinstance(row,dict) or not ident(row.get('company_id')) or not ident(row.get('id')) or type(row.get('tick')) is not int or not 0<=row['tick']<=9007199254740991 or type(row.get('quantity')) is not int or not 1<=row['quantity']<=99 or row.get('status') not in ['draft','sent']:raise ValueError('Invalid synthetic record')
  identity=(row['company_id'],row['id'])
  if identity in seen:raise ValueError('Duplicate scoped record')
  seen.add(identity)
  if row['company_id']!=query['company_id'] or status!='all' and row['status']!=status or q not in row['id'].casefold():continue
  selected.append((row['tick'],row['id'],row['quantity'],row['status']))
 rows=tuple(sorted(selected,key=lambda r:(r[0],r[1])))
 fingerprint=hashlib.sha256(packed([query['company_id'],status,q,revision,rows]).encode('utf-8')).hexdigest()
 return Snapshot(query['company_id'],status,q,revision,fingerprint,rows)

def export_rows(snapshot):return [{'id':r[1],'tick':r[0],'quantity':r[2],'status':r[3]} for r in snapshot.rows]
def summary(snapshot):return {'count':len(snapshot.rows),'quantity':sum(r[2] for r in snapshot.rows),'revision':snapshot.revision,'snapshot':snapshot.fingerprint}
def page(snapshot,limit=2,cursor=None):
 if type(limit) is not int or not 1<=limit<=50:raise ValueError('Page limit')
 start=0
 if cursor is not None:
  if not isinstance(cursor,str) or len(cursor)>500:raise ValueError('Invalid cursor')
  try:
   value=json.loads(base64.b64decode(cursor+'='*((-len(cursor))%4),altchars=b'-_',validate=True).decode('utf-8'))
  except (ValueError,UnicodeError):raise ValueError('Invalid cursor') from None
  if not isinstance(value,dict) or set(value)!= {'snapshot','after'} or value['snapshot']!=snapshot.fingerprint or not isinstance(value['after'],list) or len(value['after'])!=2:raise ValueError('Cursor snapshot mismatch')
  key=value['after']
  if type(key[0]) is not int or not ident(key[1]):raise ValueError('Invalid cursor key')
  matches=[i for i,r in enumerate(snapshot.rows) if (r[0],r[1])==tuple(key)]
  if not matches:raise ValueError('Unknown cursor position')
  start=matches[0]+1
 chunk=snapshot.rows[start:start+limit];next_cursor=None
 if chunk and start+len(chunk)<len(snapshot.rows):next_cursor=base64.urlsafe_b64encode(packed({'snapshot':snapshot.fingerprint,'after':list(chunk[-1][:2])}).encode('utf-8')).decode('ascii').rstrip('=')
 return {'rows':[{'id':r[1],'tick':r[0],'quantity':r[2],'status':r[3]} for r in chunk],'next_cursor':next_cursor,'revision':snapshot.revision,'snapshot':snapshot.fingerprint}

def synthetic_records():return [{'company_id':'A','id':'S01','tick':1,'quantity':6,'status':'draft'},{'company_id':'A','id':'S02','tick':1,'quantity':4,'status':'sent'},{'company_id':'A','id':'S03','tick':2,'quantity':3,'status':'draft'},{'company_id':'B','id':'S01','tick':1,'quantity':9,'status':'draft'}]