:j+dZddlZddlZddlZddlZddlmZmZddlm Z m Z m Z m Z ddl mZddlmZddlmZdd lmZdd lmZed ZGd d eZGddeZGddeZGddZy)z Persistence strategies for different storage modes. This module defines the strategy pattern implementation for handling different persistence approaches: simple file-based and shard-based storage. N)ABCabstractmethod)AnyDictListOptional)PersistenceMetrics) get_logger)AtomicFileWriter)NDJSONShardWriter)SnapshotManagerstoragec zeZdZdZdedeeeffdZededee de dd fd Z ede fd Z ed d Zy )PersistenceStrategyzAbstract base class for persistence strategies. Defines the interface that all persistence strategies must implement. This allows for clean separation between simple file-based and shard-based persistence approaches. directoryfilesc ||_||_y)zInitialize strategy with directory and file mappings. Args: directory: Base directory for storage files: Mapping of result types to file paths N)rr)selfrrs 9/root/.openclaw/workspace/harvester/storage/strategies.py__init__zPersistenceStrategy.__init__#s# result_typeitemsstatsreturnNcy)zWrite data items to storage. Args: result_type: Type of result being stored items: List of items to store stats: Metrics object to update N)rrrrs r write_datazPersistenceStrategy.write_data-s rcy)zCheck if this strategy supports snapshot generation. Returns: True if snapshots are supported, False otherwise Nrrs rsupports_snapshotsz&PersistenceStrategy.supports_snapshots8s rcy)zCleanup strategy resources.Nrr!s rcleanupzPersistenceStrategy.cleanupAs rrN)__name__ __module__ __qualname____doc__strrrrrrr rboolr"r$rrrrrs#d38n c $s) DV [_   D    rrcBeZdZdZdedeededdfdZde fdZ d d Z y) SimpleFileStrategyzSimple text file persistence strategy. Stores data as plain text files, one item per line. Does not support snapshots or advanced features. rrrrNc|sy|jj|}|stjd|y g}|D]H}t |dr |j |j /|j t|Jtj||tjdt|d|dy#t$r%}tjd|d|Yd}~yd}~wwxYw) zWrite items to simple text file. Args: result_type: Type of result being stored items: List of items to store stats: Metrics object to update Nz)No file path configured for result type: serializeSaved  z items to simple fileFailed to write z to simple file: ) rgetloggererrorhasattrappendr/r*r append_atomicinfolen Exception)rrrrfilepathlinesitemes rrzSimpleFileStrategy.write_dataNs ::>>+. LLD[MR S  OE ,4-LL!12LLT+  ,  * *8U ; KK&U Ak]:OP Q O LL+K=8I!M N N OsB C C4C//C4cy)z&Simple files do not support snapshots.Frr!s rr"z%SimpleFileStrategy.supports_snapshotsosrcy)z#No cleanup needed for simple files.Nrr!s rr$zSimpleFileStrategy.cleanupss rr%) r&r'r(r)r*rrr rr+r"r$rrrr-r-GsE OcO$s)ODVO[_OBD rr-cxeZdZdZdedeeefffd Zdedeede dd fd Z de fd Z dd Z dedefd ZxZS) ShardStrategyzNDJSON shard persistence strategy. Stores data as NDJSON shard files with indexing and rotation. Supports snapshot generation and advanced features. rrcft|||i|_tj|_y)zInitialize shard strategy. Args: directory: Base directory for storage files: Mapping of result types to file paths N)superr_shard_writers threadingLock_lock)rrr __class__s rrzShardStrategy.__init__s* E*<>^^% rrrrrNc|sy g}|D]`}t|dr6|j} |jtj|E|jdt |ib|j|}|j||tjdt|d|dy#t $r|jd|iYwxYw#t $r%}tjd|d|Yd}~yd}~wwxYw) zWrite items to NDJSON shard files. Args: result_type: Type of result being stored items: List of items to store stats: Metrics object to update Nr/valuer0r1z items to shardr2z to shard: ) r6r/r7jsonloadsr;r*_get_shard_writerappend_recordsr4r9r:r5) rrrrrecordsr> serializedwriterr?s rrzShardStrategy.write_datas  IG 94-!%!1J>tzz*'=> NNGSY#78 9++K8F  ! !'5 1 KK&Wa }OL M%>'<=> I LL+K= A3G H H Is;#C$B5 A'C5CCCC D DDcy)zShard files support snapshots.Trr!s rr"z ShardStrategy.supports_snapshotssrcz|j5|jjdddy#1swYyxYw)z$Cleanup shard writers and resources.N)rIrFclearr!s rr$zShardStrategy.cleanups0 ZZ (    % % ' ( ( (s1:c|j5|jj|}|sEtjj |j d}t||}||j|<|cdddS#1swYyxYw)zGet or create shard writer for result type. Args: result_type: Type of result Returns: NDJSONShardWriter instance shardsN)rIrFr3ospathjoinrr )rrrS shard_roots rrOzShardStrategy._get_shard_writerspZZ ((,,[9FWW\\$..(C *:{C39##K0    s A$A;;Br%)r&r'r(r)r*rrrrr rr+r"r$r rO __classcell__)rJs@rrCrCxsu &# &d38n &!Ic!I$s)!IDV!I[_!IFD( S5FrrCceZdZdZdedeedefdZdedefdZde eeffd Z dd edd fd Z dd Z de ee ffdZd edd fdZy )rzDedicated snapshot management with periodic building support. Manages the lifecycle of snapshot generation including periodic background building and proper cleanup. r result_types provider_namec||_||_||_d|_d|_t j |_ddddd|_y)zInitialize snapshot manager. Args: directory: Base directory containing shards result_types: List of result types to manage provider_name: Name of the provider for logging NFgr) last_snapshotsnapshot_counttotal_snapshot_timesnapshot_operations) rr_r`_periodic_thread_runningrGrHrIr)rrr_r`s rrzSnapshotManager.__init__sO#(*=A ^^% (+aX[tuv rrrc tjj|jd|}tjj |s&t j d|d|jytjj|jd}tj|dtjj||d} t||}tj}|j}tj|z }|j5tj|jd <|jd xxd z cc<|jd xx|z cc<|jd xxd z cc<dddt jd|d|d|dd|S#1swY+xYw#t$r%} t j!d|d| Yd} ~ yd} ~ wwxYw)zBuild snapshot for specific result type. Args: result_type: Type of result to build snapshot for Returns: Number of records in the snapshot rXzNo shard directory for z in r snapshotsT)exist_okz.jsonrbrcr rdreNzBuilt snapshot for z with z records in z.3fsFailed to build snapshot for : )rYrZr[risdirr4debugr`makedirsBaseSnapshotManagertimebuild_snapshotrIrr9r;r5) rrr\ snapshots_dir snapshot_pathmanager start_timecountdurationr?s rrszSnapshotManager.build_snapshotsWW\\$..(KH ww}}Z( LL2;-tDDVDVCWX Y T^^[A  MD1 ]{m54IJ  )*mDGJ**,Eyy{Z/H 7.2iik ?+ +,1, 01X=1 01Q61  7 KK-k]&|T\]`Saabc dL 7 7  LL8 RsK L s2AG)A'F:)G:G?G G4G//G4ci}|jD]} |j|||<|S#t$r*}tj d|d|d||<Yd}~Jd}~wwxYw)z}Build snapshots for all result types. Returns: Dictionary mapping result types to record counts rlrmrN)r_rsr;r4r5)rresultsrr?s rbuild_all_snapshotsz#SnapshotManager.build_all_snapshots su ,, )K )'+':':;'G $ )  ) <[MA3OP'( $ )s+ A AA interval_secNc|j5|jrF|jjr,tj d|j  dddyd|_tj|j|fdd|j |_|jjtjd|j d|ddddy#1swYyxYw) zyStart periodic snapshot building. Args: interval_sec: Interval between snapshots in seconds z&Periodic snapshot already running for NTz snapshot-)targetargsdaemonnamezStarted periodic snapshots for z (interval: zs)) rIrfis_aliver4ror`rgrGThread_periodic_loopstartr9)rr}s rstart_periodiczSnapshotManager.start_periodics ZZ l$$)>)>)G)G)I EdFXFXEYZ[ l l !DM$-$4$4**,V_`d`r`r_sTt%D !  ! ! ' ' ) KK9$:L:L9M\ZfYggij k l l lsA C& A=C&&C/c|j5|js dddyd|_|jrs|jjrY|jj d|jjr#t j d|jdd|_t jd|jdddy#1swYyxYw)z,Stop periodic snapshot building and cleanup.NFg@)timeoutzSnapshot thread for z did not stop gracefullyzStopped periodic snapshots for ) rIrgrfrr[r4warningr`r9r!s rstopzSnapshotManager.stop*s ZZ P== P P"DM$$)>)>)G)G)I%%**3*7((113NN%9$:L:L9MMe#fg$(D ! KK9$:L:L9MN O P P PsCB/CC%cz|j5|jjcdddS#1swYyxYw)zcGet snapshot statistics. Returns: Dictionary with snapshot statistics N)rIrcopyr!s r get_statszSnapshotManager.get_stats;s. ZZ %::??$ % % %s1:c|jrA tj||jsy|j|jr@yy#t$r/}t j d|jd|Yd}~Bd}~wwxYw)zmPeriodic snapshot building loop. Args: interval_sec: Interval between snapshots zError in periodic snapshot for rmN)rgrrsleepr|r;r4r5r`)rr}r?s rrzSnapshotManager._periodic_loopDsw mm Z <(}}((* mm Z >t?Q?Q>RRTUVTWXYY Zs!AA B%BB)i,r%)r&r'r(r)r*rrintrsrr|rrrrrrrrrrs w#wT#Ywsw($#$#$L T#s(^ l3ll$P"%4S>%Z3Z4Zrr)r)rMrYrGrrabcrrtypingrrrr state.modelsr tools.loggerr atomicr shardr snapshotrrqr4rr-rCrrrrss  #,,+#$$< I ) #) X. ,. bM'M`KZKZr