breakpointsoftware/document-parser

1from __future__ import annotations2 3import argparse4import json5import logging6import os7import shutil8from datetime import datetime, timezone9from pathlib import Path10from typing import Any11 12from dotenv import load_dotenv13from openai import OpenAI14 15from document_processing import extract_text, is_image_document, to_data_uri16from firebase_processed_files import (17	build_firebase_tracker,18	check_drive_documents_to_process,19)20from firebase_tenant_config import (21	FirebaseTenantConfigManager,22	CredentialsObject,23	RuleObject,24	TenantConfig,25)26from google_drive_service import (27	DRIVE_FULL_SCOPE,28	GoogleDriveConfigError,29	build_drive_service,30	move_file_to_path,31	scan_drive_supported_documents,32)33from google_sheets_service import GoogleSheetsConfigError, append_row_to_google_sheet34from receipt_ai import extract_receipt_json, extract_receipt_json_from_image, extract_receipt_json_from_pdf, extract_receipt_json_from_document35from receipt_results import build_empty_result36 37 38load_dotenv()39 40 41logger = logging.getLogger(__name__)42if not logger.handlers:43	handler = logging.StreamHandler()44	handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))45	logger.addHandler(handler)46logger.setLevel(logging.DEBUG)47logger.propagate = False48 49 50SHEET_COLUMNS = ["fecha", "source_file", "display_description", "total"]51CORRUPTED_SHEET_NAME = "corrupted data"52 53 54def _build_display_description(parsed: dict[str, Any]) -> str:55	provider_name = str(parsed.get("description_proveedor") or "").strip()56	provider_cuit = str(parsed.get("cuit_proveedor") or "").strip()57	city = str(parsed.get("ciudad") or parsed.get("city") or "").strip()58	parts = [part for part in [provider_name, provider_cuit, city] if part]59	return " - ".join(parts)60 61 62def _build_sheet_row(parsed: dict[str, Any]) -> dict[str, Any]:63	return {64		"fecha": parsed.get("fecha"),65		"source_file": parsed.get("source_file"),66		"display_description": _build_display_description(parsed),67		"total": parsed.get("total"),68	}69 70 71def _is_blank_sheet_value(value: Any) -> bool:72	if value is None:73		return True74 75	if isinstance(value, str):76		return not value.strip()77 78	return False79 80 81def _is_complete_sheet_row(row: dict[str, Any]) -> bool:82	return all(not _is_blank_sheet_value(row.get(column)) for column in SHEET_COLUMNS)83 84 85def _create_openai_client_from_key(api_key: str) -> OpenAI:86	"""Create OpenAI client from API key."""87	if not api_key.strip():88		raise RuntimeError("Missing OPENAI_API_KEY.")89	return OpenAI(api_key=api_key)90 91 92def _parse_invoice_date(value: Any) -> datetime | None:93	text = str(value or "").strip()94	if not text:95		return None96 97	for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%Y/%m/%d"):98		try:99			return datetime.strptime(text, fmt)100		except ValueError:101			continue102 103	try:104		return datetime.fromisoformat(text.replace("Z", "+00:00"))105	except ValueError:106		return None107 108 109def _resolve_drive_destination_path(parsed: dict[str, Any]) -> tuple[str, bool]:110	base_path = (os.getenv("GOOGLE_DRIVE_INVOICES_BASE_PATH") or "Facturas").strip().strip("/") or "Facturas"111 112	row_for_sheet = _build_sheet_row(parsed)113	if not _is_complete_sheet_row(row_for_sheet):114		return f"Corrupted", True115 116	invoice_date = _parse_invoice_date(parsed.get("fecha"))117	if invoice_date is None:118		return f"Corrupted", True119 120	return f"{invoice_date.strftime('%Y%m')}", False121 122 123def _build_drive_file_url(file_id: str) -> str:124	return f"https://drive.google.com/file/d/{file_id}/view"125 126 127def _build_hyperlink_formula(url: str, label: str) -> str:128	safe_url = str(url).replace('"', '""')129	safe_label = str(label).replace('"', '""')130	return f'=HYPERLINK("{safe_url}","{safe_label}")'131 132 133def _parse_document(client: OpenAI, model: str, local_path: str, source_file: str) -> dict[str, Any]:134	path = Path(local_path)135	136	# Debug: Check if file exists137	if not path.exists():138		logger.error("Document file does not exist: %s", local_path)139		return build_empty_result(source_file)140	141	logger.debug("Parsing document: %s (size: %s bytes)", local_path, path.stat().st_size)142	143	try:144		# Send all documents in their original format to OpenAI145		file_type = path.suffix.lower()146		147		if file_type in {".jpg", ".jpeg", ".png"}:148			# Send image in original format149			logger.debug("Sending image in original format to OpenAI")150			parsed = extract_receipt_json_from_image(client, model, path, to_data_uri(path))151		152		elif file_type == ".pdf":153			# Send PDF in original format154			logger.debug("Sending PDF in original format to OpenAI")155			parsed = extract_receipt_json_from_pdf(client, model, path)156		157		else:158			# Send other document types (TXT, DOCX, etc) in original format159			logger.debug("Sending %s in original format to OpenAI", file_type)160			parsed = extract_receipt_json_from_document(client, model, path)161		162		parsed["source_file"] = source_file163		return parsed164	165	except Exception as exc:166		logger.exception("Error parsing document %s: %s", local_path, exc)167		parsed = build_empty_result(source_file)168		parsed["source_file"] = source_file169		return parsed170 171 172def _create_default_tenant_from_env() -> tuple[TenantConfig, RuleObject] | None:173	"""Create a default tenant from environment variables if possible.174	175	Maps legacy env vars to a default multi-tenant configuration:176	- OPENAI_API_KEY -> credentials.openai_api_key177	- GOOGLE_SERVICE_ACCOUNT_JSON -> credentials.google_service_account_json178	- GOOGLE_DRIVE_FOLDER_IDS -> rule.source_folder_id (first one)179	- GOOGLE_DRIVE_INVOICES_ROOT_FOLDER_ID -> rule.target_folder_id180	- GOOGLE_SHEETS_SPREADSHEET_ID -> rule.target_sheet_id181	182	Returns:183		TenantConfig if all required env vars are set, None otherwise184	"""185	openai_api_key = (os.getenv("OPENAI_API_KEY") or "").strip()186	google_service_account_json = (os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON") or "").strip()187	drive_folder_ids = (os.getenv("GOOGLE_DRIVE_FOLDER_IDS") or "").strip()188	target_folder_id = (os.getenv("GOOGLE_DRIVE_INVOICES_ROOT_FOLDER_ID") or "").strip()189	target_sheet_id = (os.getenv("GOOGLE_SHEETS_SPREADSHEET_ID") or "").strip()190	sheet_tab_name = (os.getenv("GOOGLE_SHEETS_WORKSHEET") or "Salidas").strip()191	192	# Check if all required vars are set193	if not all([openai_api_key, google_service_account_json, drive_folder_ids, target_folder_id, target_sheet_id]):194		logger.warning(195			"Cannot create default tenant: missing env vars. "196			"Please ensure OPENAI_API_KEY, GOOGLE_SERVICE_ACCOUNT_JSON, GOOGLE_DRIVE_FOLDER_IDS, "197			"GOOGLE_DRIVE_INVOICES_ROOT_FOLDER_ID, and GOOGLE_SHEETS_SPREADSHEET_ID are set."198		)199		return None200	201	# Get first folder ID from comma-separated list202	source_folder_id = drive_folder_ids.split(",")[0].strip()203	204	# Create credentials205	credentials = CredentialsObject(206		openai_api_key=openai_api_key,207		google_service_account_json=google_service_account_json,208	)209	210	# Create default rule211	default_rule = RuleObject(212		rule_id="default_rule",213		rule_name="Default Processing Rule",214		source_folder_id=source_folder_id,215		target_folder_id=target_folder_id,216		target_sheet_id=target_sheet_id,217		sheet_tab_name=sheet_tab_name,218		parsing_prompt=None,219		is_enabled=True,220	)221	222	# Create tenant (rules will be added as subcollection)223	tenant = TenantConfig(224		tenant_id="default_tenant",225		name="Default Tenant (from .env)",226		active=True,227		credentials=credentials,228		created_at=datetime.now(timezone.utc).isoformat(),229	)230	231	logger.info("Created default tenant from environment variables")232	return tenant, default_rule233 234 235def orchestrate_single_rule(236	tenant_id: str,237	tenant_config: TenantConfig,238	rule: RuleObject,239	model: str,240	include_subfolders: bool = True,241	send_to_sheet: bool = True,242) -> dict[str, Any]:243	"""Orchestrate document processing for a single rule of a single tenant.244	245	Args:246		tenant_id: Unique tenant identifier247		tenant_config: Tenant configuration (with credentials and rules)248		rule: The processing rule to orchestrate249		model: OpenAI model to use for parsing250		include_subfolders: Whether to include subfolders when scanning Drive251		send_to_sheet: Whether to append results to Google Sheets252	253	Returns:254		Summary dict with processing results255	"""256	logger.info(257		"Starting orchestration for tenant=%s rule=%s model=%s",258		tenant_id,259		rule.rule_id,260		model,261	)262	263	# Validate tenant credentials264	if not tenant_config.credentials.openai_api_key:265		logger.error("Tenant=%s rule=%s missing openai_api_key", tenant_id, rule.rule_id)266		return {267			"ok": False,268			"error": f"Tenant {tenant_id} missing OpenAI API key",269			"tenant_id": tenant_id,270			"rule_id": rule.rule_id,271		}272	273	if not tenant_config.credentials.google_service_account_json:274		logger.error("Tenant=%s rule=%s missing google_service_account_json", tenant_id, rule.rule_id)275		return {276			"ok": False,277			"error": f"Tenant {tenant_id} missing Google service account credentials",278			"tenant_id": tenant_id,279			"rule_id": rule.rule_id,280		}281	282	# Set up clients for this tenant283	try:284		client = _create_openai_client_from_key(tenant_config.credentials.openai_api_key)285		tracker = build_firebase_tracker(tenant_id=tenant_id, rule_id=rule.rule_id)286	except RuntimeError as exc:287		logger.error("Tenant=%s rule=%s failed to initialize clients: %s", tenant_id, rule.rule_id, exc)288		return {289			"ok": False,290			"error": str(exc),291			"tenant_id": tenant_id,292			"rule_id": rule.rule_id,293		}294	295	# Set up Google Drive service with tenant's credentials296	scan_temp_dir: str | None = None297	try:298		# Build drive service with tenant credentials299		os.environ["GOOGLE_SERVICE_ACCOUNT_JSON"] = tenant_config.credentials.google_service_account_json300		drive_service = build_drive_service(scope=DRIVE_FULL_SCOPE)301		302		# Scan the rule's source folder303		scan_result = scan_drive_supported_documents(304			folder_ids=[rule.source_folder_id],305			include_subfolders=include_subfolders,306		)307		drive_documents = scan_result.documents308		scan_temp_dir = scan_result.temp_dir309	except GoogleDriveConfigError as exc:310		logger.error("Tenant=%s rule=%s Google Drive config error: %s", tenant_id, rule.rule_id, exc)311		return {312			"ok": False,313			"error": f"Google Drive configuration error: {exc}",314			"tenant_id": tenant_id,315			"rule_id": rule.rule_id,316		}317	318	logger.info(319		"Tenant=%s rule=%s scan completed documents_found=%s",320		tenant_id,321		rule.rule_id,322		len(drive_documents),323	)324	325	# Prepare docs payload326	docs_payload = [327		{328			"document_id": doc.document_id,329			"source_file": doc.source_file,330			"modificationDate": doc.modificationDate,331			"local_path": doc.local_path,332		}333		for doc in drive_documents334	]335	336	# Check which documents need processing337	to_process, skipped, tracking_warning = check_drive_documents_to_process(tracker, docs_payload)338	logger.info(339		"Tenant=%s rule=%s to_process=%s skipped=%s",340		tenant_id,341		rule.rule_id,342		len(to_process),343		len(skipped),344	)345	if tracking_warning:346		logger.warning("Tenant=%s rule=%s tracking warning: %s", tenant_id, rule.rule_id, tracking_warning)347	348	# Process documents349	parsed_count = 0350	modified_count = 0351	sent_count = 0352	moved_count = 0353	corrupted_count = 0354	error_items: list[dict[str, str]] = []355	processed_items: list[dict[str, Any]] = []356	357	for document in to_process:358		document_id = str(document.get("document_id") or "").strip()359		source_file = str(document.get("source_file") or "").strip()360		local_path = str(document.get("local_path") or "").strip()361		modification_date = document.get("modificationDate")362		status = str(document.get("target_status") or "Parsed")363		file_hash = str(document.get("file_hash") or "").strip() or None364		365		logger.info(366			"Tenant=%s rule=%s processing document_id=%s source_file=%s",367			tenant_id,368			rule.rule_id,369			document_id,370			source_file,371		)372		373		if tracker.is_processed(file_hash):374			logger.info(375				"Tenant=%s rule=%s skipping document_id=%s already processed",376				tenant_id,377				rule.rule_id,378				document_id,379			)380			skipped.append({"document_id": document_id, "source_file": source_file, "reason": "Already processed"})381			continue382		383		try:384			parsed = _parse_document(client, model, local_path, source_file)385			logger.info(386				"Tenant=%s rule=%s parsed document_id=%s",387				tenant_id,388				rule.rule_id,389				document_id,390			)391			392			# Determine destination and whether corrupted393			destination_path, is_corrupted = _resolve_drive_destination_path(parsed)394			if is_corrupted:395				corrupted_count += 1396			397			# Move file to destination (if target_folder_id is set)398			if rule.target_folder_id:399				move_file_to_path(400					service=drive_service,401					file_id=document_id,402					destination_path=destination_path,403					root_folder_id=rule.target_folder_id,404				)405				moved_count += 1406				logger.info(407					"Tenant=%s rule=%s moved document_id=%s to %s",408					tenant_id,409					rule.rule_id,410					document_id,411					destination_path,412				)413			414			# Save document record415			if tracker is not None:416				tracker.save_document_record(417					file_hash=file_hash,418					document_id=document_id,419					source_file=source_file,420					modification_date=modification_date,421					status=status,422					parsed_data=parsed,423				)424			425			if status == "Modified":426				modified_count += 1427			else:428				parsed_count += 1429			430			# Send to Google Sheets if enabled431			worksheet_name: str | None = None432			if rule.target_sheet_id:433				os.environ["GOOGLE_SHEETS_SPREADSHEET_ID"] = rule.target_sheet_id434				row_for_sheet = _build_sheet_row(parsed)435				row_for_sheet["source_file"] = _build_hyperlink_formula(436					_build_drive_file_url(document_id),437					source_file,438				)439				worksheet_name = "Corrupted_data" if not _is_complete_sheet_row(row_for_sheet) else rule.sheet_tab_name440				441				try:442					append_row_to_google_sheet(row_for_sheet, SHEET_COLUMNS, worksheet_name=worksheet_name)443					logger.info(444						"Tenant=%s rule=%s sent document_id=%s to sheet=%s",445						tenant_id,446						rule.rule_id,447						document_id,448						worksheet_name,449					)450					if tracker is not None and file_hash:451						tracker.mark_document_sent(file_hash)452					sent_count += 1453				except GoogleSheetsConfigError as exc:454					logger.error(455						"Tenant=%s rule=%s sheet error for document_id=%s: %s",456						tenant_id,457						rule.rule_id,458						document_id,459						exc,460					)461					error_items.append({462						"document_id": document_id,463						"source_file": source_file,464						"error": f"Google Sheets error: {exc}",465					})466					continue467			468			processed_items.append({469				"document_id": document_id,470				"source_file": source_file,471				"modificationDate": modification_date,472				"status": "Corrupted" if is_corrupted else ("Sent" if send_to_sheet else status),473				"destination_path": destination_path,474			})475		476		except Exception as exc:477			logger.exception(478				"Tenant=%s rule=%s failed processing document_id=%s",479				tenant_id,480				rule.rule_id,481				document_id,482			)483			error_items.append({484				"document_id": document_id,485				"source_file": source_file,486				"error": str(exc),487			})488	489	# Cleanup490	if scan_temp_dir:491		shutil.rmtree(scan_temp_dir, ignore_errors=True)492		logger.info("Tenant=%s rule=%s cleaned up temp directory", tenant_id, rule.rule_id)493	494	logger.info(495		"Tenant=%s rule=%s orchestration finished parsed=%s modified=%s sent=%s moved=%s corrupted=%s errors=%s",496		tenant_id,497		rule.rule_id,498		parsed_count,499		modified_count,500		sent_count,501		moved_count,502		corrupted_count,503		len(error_items),504	)505	506	return {507		"ok": True,508		"tenant_id": tenant_id,509		"rule_id": rule.rule_id,510		"rule_name": rule.rule_name,511		"scanned": len(drive_documents),512		"to_process": len(to_process),513		"skipped": len(skipped),514		"parsed": parsed_count,515		"modified": modified_count,516		"sent": sent_count,517		"moved": moved_count,518		"corrupted": corrupted_count,519		"tracking_warning": tracking_warning,520		"processed_items": processed_items,521		"skipped_items": skipped,522		"errors": error_items,523	}524 525 526def orchestrate_all_active_tenants(527	model: str,528	include_subfolders: bool = True,529	send_to_sheet: bool = True,530) -> dict[str, Any]:531	"""Orchestrate all active tenants and their enabled rules.532	533	This is the main multi-tenant entry point that:534	1. Loads all active tenants from Firebase535	2. For each tenant, processes all enabled rules536	3. Aggregates results across all tenants537	538	Args:539		model: OpenAI model to use for parsing540		include_subfolders: Whether to include subfolders when scanning Drive541		send_to_sheet: Whether to append results to Google Sheets542	543	Returns:544		Summary dict with aggregated processing results545	"""546	logger.info(547		"Starting multi-tenant orchestration model=%s include_subfolders=%s send_to_sheet=%s",548		model,549		include_subfolders,550		send_to_sheet,551	)552	553	# Load all active tenants from Firebase554	config_manager = FirebaseTenantConfigManager()555	default_tenant_created = False556	default_tenant_saved = False557	558	try:559		active_tenants = config_manager.list_active_tenants()560	except Exception as exc:561		logger.error("Failed to load active tenants: %s", exc)562		return {563			"ok": False,564			"error": f"Failed to load tenants: {exc}",565			"tenants_processed": 0,566		}567	568	logger.info("Loaded %s active tenants", len(active_tenants))569	570	# If no tenants found, create and save default tenant BEFORE orchestration571	if not active_tenants:572		logger.warning("No active tenants found. Creating default tenant from .env...")573		result = _create_default_tenant_from_env()574		575		if not result:576			return {577				"ok": False,578				"error": "No active tenants found in Firebase and could not create default tenant from .env",579				"tenants_processed": 0,580				"tenants_results": [],581			}582		583		default_tenant, default_rule = result584		585		# Save default tenant to Firebase BEFORE orchestration586		try:587			logger.info("Saving default tenant '%s' to Firebase...", default_tenant.tenant_id)588			config_manager.save_tenant(default_tenant)589			# Add the default rule to the tenant's rules subcollection590			config_manager.add_rule(default_tenant.tenant_id, default_rule)591			logger.info("✓ Successfully saved default tenant and rule to Firebase: %s", default_tenant.tenant_id)592			default_tenant_saved = True593		except Exception as exc:594			logger.error("✗ Failed to save default tenant to Firebase: %s", exc, exc_info=True)595			return {596				"ok": False,597				"error": f"Failed to save default tenant to Firebase: {exc}",598				"tenants_processed": 0,599				"tenants_results": [],600			}601		602		# Use the default tenant for orchestration603		active_tenants = [default_tenant]604		default_tenant_created = True605		logger.info("Using default tenant created and saved from environment variables")606	607	# Process each tenant and their rules608	tenants_results = []609	total_parsed = 0610	total_modified = 0611	total_sent = 0612	total_moved = 0613	total_corrupted = 0614	total_errors = 0615	616	for tenant in active_tenants:617		tenant_id = tenant.tenant_id618		logger.info("Processing tenant=%s name=%s", tenant_id, tenant.name)619		620		# Get enabled rules for this tenant from the rules subcollection621		enabled_rules = config_manager.get_enabled_rules(tenant_id)622		logger.info("Tenant=%s has %s enabled rules", tenant_id, len(enabled_rules))623		624		if not enabled_rules:625			logger.warning("Tenant=%s has no enabled rules", tenant_id)626			tenants_results.append({627				"tenant_id": tenant_id,628				"tenant_name": tenant.name,629				"ok": True,630				"warning": "No enabled rules",631				"rules_processed": 0,632				"rules_results": [],633			})634			continue635		636		# Process each rule637		rules_results = []638		tenant_parsed = 0639		tenant_modified = 0640		tenant_sent = 0641		tenant_moved = 0642		tenant_corrupted = 0643		tenant_errors = 0644		645		for rule in enabled_rules:646			rule_result = orchestrate_single_rule(647				tenant_id=tenant_id,648				tenant_config=tenant,649				rule=rule,650				model=model,651				include_subfolders=include_subfolders,652				send_to_sheet=send_to_sheet,653			)654			rules_results.append(rule_result)655			656			if rule_result.get("ok"):657				tenant_parsed += rule_result.get("parsed", 0)658				tenant_modified += rule_result.get("modified", 0)659				tenant_sent += rule_result.get("sent", 0)660				tenant_moved += rule_result.get("moved", 0)661				tenant_corrupted += rule_result.get("corrupted", 0)662				tenant_errors += len(rule_result.get("errors", []))663		664		total_parsed += tenant_parsed665		total_modified += tenant_modified666		total_sent += tenant_sent667		total_moved += tenant_moved668		total_corrupted += tenant_corrupted669		total_errors += tenant_errors670		671		tenants_results.append({672			"tenant_id": tenant_id,673			"tenant_name": tenant.name,674			"ok": True,675			"rules_processed": len(rules_results),676			"parsed": tenant_parsed,677			"modified": tenant_modified,678			"sent": tenant_sent,679			"moved": tenant_moved,680			"corrupted": tenant_corrupted,681			"errors": tenant_errors,682			"rules_results": rules_results,683		})684	685	logger.info(686		"Multi-tenant orchestration completed tenants=%s parsed=%s modified=%s sent=%s moved=%s corrupted=%s errors=%s",687		len(tenants_results),688		total_parsed,689		total_modified,690		total_sent,691		total_moved,692		total_corrupted,693		total_errors,694	)695	696	return {697		"ok": True,698		"tenants_processed": len(tenants_results),699		"default_tenant_created": default_tenant_created,700		"default_tenant_saved": default_tenant_saved,701		"total_parsed": total_parsed,702		"total_modified": total_modified,703		"total_sent": total_sent,704		"total_moved": total_moved,705		"total_corrupted": total_corrupted,706		"total_errors": total_errors,707		"tenants_results": tenants_results,708	}709 710 711def parse_args() -> argparse.Namespace:712	parser = argparse.ArgumentParser(713		description="Orchestrate Drive documents: scan, parse, and persist status using multi-tenant Firebase configuration."714	)715	parser.add_argument(716		"--model",717		default=os.getenv("OPENAI_MODEL", "gpt-4o"),718		help="OpenAI model name."719	)720	parser.add_argument(721		"--no-subfolders",722		action="store_true",723		help="Only scan direct files in configured Drive folders.",724	)725	parser.add_argument(726		"--send",727		action="store_true",728		help="After parsing, append each row to Google Sheets and mark status as Sent.",729	)730	parser.add_argument(731		"--output",732		default="",733		help="Optional output JSON path for the orchestration summary.",734	)735	return parser.parse_args()736 737 738def main() -> int:739	args = parse_args()740	logger.info(741		"Running document orchestration CLI in multi-tenant mode output=%s",742		args.output or "stdout",743	)744	745	logger.info("Loading all active tenants from Firebase")746	summary = orchestrate_all_active_tenants(747		model=args.model,748		include_subfolders=not args.no_subfolders,749		send_to_sheet=args.send,750	)751 752	rendered = json.dumps(summary, indent=2, ensure_ascii=False)753	if args.output:754		output_path = Path(args.output)755		output_path.parent.mkdir(parents=True, exist_ok=True)756		output_path.write_text(rendered, encoding="utf-8")757		logger.info("Wrote orchestration summary to %s", output_path)758	else:759		logger.info("Orchestration summary:\n%s", rendered)760 761	return 0 if summary.get("ok") else 1762 763 764if __name__ == "__main__":765	raise SystemExit(main())766