import os from datetime import datetime, timedelta from pypos.core.base_model import BaseModel from pypos.core.utils.db_helper import connect_sqlite from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.sql_query_builder import build_sql_with_identifier_in_clause from pypos.core.utils.sql_identifier_utils import quote_sql_identifier from pypos.core.database.schema_migrator import run_schema_migrations_once # edited by glg class ExportUploadModel(BaseModel): def __init__(self, db_path=None): super().__init__() self.db_path = db_path or get_db_path() self._schema_ready = False def _connect(self): run_schema_migrations_once(self.db_path, strict=False) conn = connect_sqlite(self.db_path) conn.row_factory = self._row_factory() self._ensure_upload_schema(conn) return conn def _row_factory(self): try: import sqlite3 return sqlite3.Row except (ImportError, ModuleNotFoundError): return None def _ensure_upload_schema(self, conn): if self._schema_ready: return cur = conn.cursor() self._ensure_column(cur, "export_flux", "upload_status", "TEXT") self._ensure_column(cur, "export_flux", "upload_attempt_count", "INTEGER") self._ensure_column(cur, "export_flux", "upload_next_retry_at", "TEXT") self._ensure_column(cur, "export_flux", "upload_error_log", "TEXT") self._ensure_column(cur, "export_flux", "uploaded_at", "TEXT") self._ensure_column(cur, "export_flux", "upload_updated_at", "TEXT") self._ensure_column(cur, "export_flux", "upload_response", "TEXT") cur.execute( "CREATE INDEX IF NOT EXISTS idx_export_flux_upload_status ON export_flux(upload_status, upload_next_retry_at)" ) conn.commit() self._schema_ready = True def _ensure_column(self, cursor, table_name, column_name, column_type): table_sql = quote_sql_identifier(table_name) column_sql = quote_sql_identifier(column_name) type_sql = str(column_type or "").strip().upper() if type_sql not in {"TEXT", "INTEGER", "REAL", "BLOB", "NUMERIC"}: raise ValueError(f"Tipe kolom tidak valid: {column_type}") cursor.execute(f"PRAGMA table_info({table_sql})") cols = {str(row[1]) for row in cursor.fetchall() if row and len(row) > 1} if column_name not in cols: cursor.execute(f"ALTER TABLE {table_sql} ADD COLUMN {column_sql} {type_sql}") def fetch_pending_uploads(self, limit): conn = self._connect() try: cur = conn.cursor() cur.execute( """ SELECT id, table_name, server_hash, file_seq, row_count, batch_start, batch_end, file_path, file_hash, file_size, status, COALESCE(upload_status, 'PENDING') AS upload_status, COALESCE(upload_attempt_count, 0) AS upload_attempt_count, COALESCE(upload_next_retry_at, '') AS upload_next_retry_at, COALESCE(upload_error_log, '') AS upload_error_log, COALESCE(uploaded_at, '') AS uploaded_at, COALESCE(created_at, '') AS created_at, COALESCE(updated_at, '') AS updated_at FROM export_flux WHERE status = 'SUCCESS' AND COALESCE(file_path, '') <> '' AND COALESCE(upload_status, 'PENDING') IN ('PENDING', 'RETRY') AND ( COALESCE(upload_next_retry_at, '') = '' OR datetime(upload_next_retry_at) <= datetime('now', 'localtime') ) ORDER BY id ASC LIMIT ? """, (int(limit),), ) rows = cur.fetchall() return [dict(row) for row in rows] finally: conn.close() def mark_uploaded(self, flux_id, response_text=""): now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() try: cur = conn.cursor() cur.execute( """ UPDATE export_flux SET upload_status = 'UPLOADED', upload_next_retry_at = NULL, upload_error_log = NULL, upload_response = ?, uploaded_at = ?, upload_updated_at = ? WHERE id = ? """, (str(response_text or "")[:2000], now, now, int(flux_id)), ) conn.commit() finally: conn.close() # edited by glg def mark_failed(self, flux_id, error_message, response_text=""): now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() try: cur = conn.cursor() cur.execute( """ UPDATE export_flux SET upload_status = 'FAILED', upload_next_retry_at = NULL, upload_error_log = ?, upload_response = ?, upload_updated_at = ?, upload_attempt_count = COALESCE(upload_attempt_count, 0) + 1 WHERE id = ? """, (str(error_message or "")[:2000], str(response_text or "")[:2000], now, int(flux_id)), ) conn.commit() finally: conn.close() def mark_retry(self, flux_id, error_message, max_attempt, backoff_base, backoff_factor, backoff_max, response_text=""): now_dt = datetime.now() now = now_dt.strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() try: cur = conn.cursor() cur.execute( "SELECT COALESCE(upload_attempt_count, 0) FROM export_flux WHERE id = ?", (int(flux_id),), ) row = cur.fetchone() current_attempt = int((row[0] if row else 0) or 0) next_attempt = current_attempt + 1 max_attempt = max(1, int(max_attempt)) if next_attempt >= max_attempt: cur.execute( """ UPDATE export_flux SET upload_status = 'FAILED', upload_next_retry_at = NULL, upload_error_log = ?, upload_response = ?, upload_updated_at = ?, upload_attempt_count = ? WHERE id = ? """, (str(error_message or "")[:2000], str(response_text or "")[:2000], now, next_attempt, int(flux_id)), ) conn.commit() return backoff_base = max(1.0, float(backoff_base)) backoff_factor = max(1.0, float(backoff_factor)) backoff_max = max(backoff_base, float(backoff_max)) wait_seconds = min(backoff_max, backoff_base * (backoff_factor ** max(0, next_attempt - 1))) next_retry_at = (now_dt + timedelta(seconds=wait_seconds)).strftime("%Y-%m-%d %H:%M:%S") cur.execute( """ UPDATE export_flux SET upload_status = 'RETRY', upload_next_retry_at = ?, upload_error_log = ?, upload_response = ?, upload_updated_at = ?, upload_attempt_count = ? WHERE id = ? """, (next_retry_at, str(error_message or "")[:2000], str(response_text or "")[:2000], now, next_attempt, int(flux_id)), ) conn.commit() finally: conn.close() # edited by glg def recover_failed_transient_rows(self, retryable_terms=None, limit=100, max_age_hours=72): terms = [] for raw in (retryable_terms or []): text = str(raw or "").strip().lower() if not text: continue if text in terms: continue terms.append(text) if not terms: return 0 limit = max(1, int(limit or 100)) max_age_hours = max(1, int(max_age_hours or 72)) now_dt = datetime.now() now = now_dt.strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() requeued = 0 try: cur = conn.cursor() cur.execute( """ SELECT id, COALESCE(upload_error_log, '') AS upload_error_log, COALESCE(upload_updated_at, updated_at, created_at, '') AS last_marked_at FROM export_flux WHERE status = 'SUCCESS' AND COALESCE(upload_status, 'PENDING') = 'FAILED' AND COALESCE(file_path, '') <> '' ORDER BY id ASC LIMIT ? """, (limit,), ) rows = cur.fetchall() or [] for row in rows: row_id = int(row[0] or 0) if row_id <= 0: continue err_text = str(row[1] or "").strip().lower() if not err_text: continue if not any(term in err_text for term in terms): continue last_marked_at_raw = str(row[2] or "").strip() if last_marked_at_raw: try: last_dt = datetime.strptime(last_marked_at_raw, "%Y-%m-%d %H:%M:%S") age_hours = max(0.0, (now_dt - last_dt).total_seconds() / 3600.0) if age_hours > float(max_age_hours): continue except (TypeError, ValueError): pass cur.execute( """ UPDATE export_flux SET upload_status = 'RETRY', upload_next_retry_at = ?, upload_updated_at = ?, upload_attempt_count = 0 WHERE id = ? """, (now, now, row_id), ) if int(cur.rowcount or 0) > 0: requeued += 1 if requeued > 0: conn.commit() else: conn.rollback() return requeued finally: conn.close() def cleanup_uploaded_files(self, retention_days, limit=100): retention_days = max(0, int(retention_days)) if retention_days <= 0: return 0 cutoff = (datetime.now() - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() deleted_count = 0 try: cur = conn.cursor() cur.execute( """ SELECT id, file_path FROM export_flux WHERE upload_status = 'UPLOADED' AND COALESCE(file_path, '') <> '' AND COALESCE(uploaded_at, '') <> '' AND datetime(uploaded_at) <= datetime(?) ORDER BY uploaded_at ASC LIMIT ? """, (cutoff, int(limit)), ) rows = cur.fetchall() for row in rows: flux_id = int(row[0]) file_path = str(row[1] or "").strip() removed = False try: if file_path and os.path.isfile(file_path): os.remove(file_path) removed = True elif file_path: removed = True except (OSError, PermissionError): removed = False if not removed: continue cur.execute( """ UPDATE export_flux SET file_path = NULL, upload_updated_at = ? WHERE id = ? """, (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), flux_id), ) deleted_count += 1 conn.commit() return deleted_count finally: conn.close() def cleanup_empty_flux_rows(self, retention_days, limit=500): retention_days = max(0, int(retention_days)) limit = max(1, int(limit)) if retention_days <= 0: return 0 cutoff = (datetime.now() - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S") conn = self._connect() deleted_count = 0 try: cur = conn.cursor() cur.execute( """ SELECT id FROM export_flux WHERE status = 'SUCCESS' AND COALESCE(file_path, '') = '' AND datetime(COALESCE(updated_at, created_at, '1970-01-01 00:00:00')) <= datetime(?) ORDER BY id ASC LIMIT ? """, (cutoff, limit), ) ids = [int(row[0]) for row in cur.fetchall() if row and row[0] is not None] if not ids: return 0 delete_query, delete_params = build_sql_with_identifier_in_clause( "DELETE FROM export_flux WHERE", "id", ids, cast_int=True, positive_only=True, unique=True, ) cur.execute(delete_query, delete_params) deleted_count = int(cur.rowcount or 0) conn.commit() return deleted_count finally: conn.close()