import os from typing import Optional, List from lib.core.texttools.texttools import name_fix from lib.data.ourtime.ourtime import OurTime, new as ourtime_new from lib.core.logger.model import Logger, LogItem, LogType class SearchArgs: def __init__(self, timestamp_from: Optional[OurTime] = None, timestamp_to: Optional[OurTime] = None, cat: str = "", log: str = "", logtype: Optional[LogType] = None, maxitems: int = 10000): self.timestamp_from = timestamp_from self.timestamp_to = timestamp_to self.cat = cat self.log = log self.logtype = logtype self.maxitems = maxitems def process(result: List[LogItem], current_item: LogItem, current_time: OurTime, args: SearchArgs, from_time: int, to_time: int): # Add previous item if it matches filters log_epoch = current_item.timestamp.unix() if log_epoch < from_time or log_epoch > to_time: return cat_match = (args.cat == '' or current_item.cat.strip() == args.cat) log_match = (args.log == '' or args.log.lower() in current_item.log.lower()) logtype_match = (args.logtype is None or current_item.logtype == args.logtype) if cat_match and log_match and logtype_match: result.append(current_item) def search(l: Logger, args_: SearchArgs) -> List[LogItem]: args = args_ # Format category (max 10 chars, ascii only) args.cat = name_fix(args.cat) if len(args.cat) > 10: raise ValueError('category cannot be longer than 10 chars') timestamp_from = args.timestamp_from if args.timestamp_from else OurTime() timestamp_to = args.timestamp_to if args.timestamp_to else OurTime() # Get time range from_time = timestamp_from.unix() to_time = timestamp_to.unix() if from_time > to_time: raise ValueError(f'from_time cannot be after to_time: {from_time} < {to_time}') result: List[LogItem] = [] # Find log files in time range files = sorted(os.listdir(l.path.path)) for file in files: if not file.endswith('.log'): continue # Parse dayhour from filename dayhour = file[:-4] # remove .log try: file_time = ourtime_new(dayhour) except ValueError: continue # Skip if filename is not a valid time format current_time = OurTime() current_item = LogItem(OurTime(), "", "", LogType.STDOUT) # Initialize with dummy values collecting = False # Skip if file is outside time range if file_time.unix() < from_time or file_time.unix() > to_time: continue # Read and parse log file content = "" try: with open(os.path.join(l.path.path, file), 'r') as f: content = f.read() except FileNotFoundError: continue lines = content.split('\n') for line in lines: if len(result) >= args.maxitems: return result line_trim = line.strip() if not line_trim: continue # Check if this is a timestamp line if not (line.startswith(' ') or line.startswith('E')): try: current_time = ourtime_new(line_trim) except ValueError: continue # Skip if not a valid timestamp line if collecting: process(result, current_item, current_time, args, from_time, to_time) collecting = False continue if collecting and len(line) > 14 and line[13] == '-': process(result, current_item, current_time, args, from_time, to_time) collecting = False # Parse log line is_error = line.startswith('E') if not collecting: # Start new item cat_start = 2 cat_end = 12 log_start = 15 if len(line) < log_start: continue # Line too short to contain log content current_item = LogItem( timestamp=current_time, cat=line[cat_start:cat_end].strip(), log=line[log_start:].strip(), logtype=LogType.ERROR if is_error else LogType.STDOUT ) collecting = True else: # Continuation line if len(line_trim) < 16: # Check for minimum length for continuation line current_item.log += '\n' + line_trim else: current_item.log += '\n' + line[15:].strip() # Use strip for continuation lines # Add last item if collecting if collecting: process(result, current_item, current_time, args, from_time, to_time) return result