Skip to content

Services API

Backup - admin.services.backup

admin.services.backup.create_backup 

create_backup(label: str = 'auto') -> str

Create a WAL-consistent snapshot of the SQLite DB.

Uses sqlite3's online backup API instead of shutil.copy2 so that any data still sitting in the -wal file is included. Do NOT swap this back to a plain file copy while journal_mode=WAL is active.

Source code in admin/services/backup.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def create_backup(label: str = "auto") -> str:
    """Create a WAL-consistent snapshot of the SQLite DB.

    Uses sqlite3's online backup API instead of shutil.copy2 so that
    any data still sitting in the -wal file is included. Do NOT swap
    this back to a plain file copy while journal_mode=WAL is active.
    """
    if not os.path.exists(DB_PATH):
        raise FileNotFoundError(f"Database not found: {DB_PATH}")

    os.makedirs(BACKUP_DIR, exist_ok=True)

    safe_label = "".join(ch for ch in str(label) if ch.isalnum() or ch in "-_") or "auto"

    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    fn = os.path.join(BACKUP_DIR, f"backup_{safe_label}_{ts}.db")

    src = sqlite3.connect(DB_PATH)
    try:
        dst = sqlite3.connect(fn)
        try:
            src.backup(dst)  # atomic, WAL-aware snapshot
            chk = dst.execute("PRAGMA quick_check").fetchone()
            if not chk or chk[0] != "ok":
                raise RuntimeError(f"Backup integrity check failed: {chk}")
        finally:
            dst.close()
    finally:
        src.close()

    return os.path.abspath(fn)

admin.services.backup.list_backups 

list_backups() -> list

Return backup filenames (newest first).

Source code in admin/services/backup.py 
57
58
59
60
61
62
63
64
def list_backups() -> list:
    """Return backup filenames (newest first)."""
    if not os.path.isdir(BACKUP_DIR):
        return []
    return sorted(
        [f for f in os.listdir(BACKUP_DIR) if f.endswith(".db")],
        reverse=True,
    )

admin.services.backup.restore_backup 

restore_backup(backup_filename: str) -> str

Atomically restore a backup into the live DB.

  • Verifies the backup file's integrity before touching anything.
  • Uses sqlite3's backup API (WAL-aware) to overwrite the live DB.
  • Removes stale -wal / -shm sidecar files.
  • Refuses to run if the backup path escapes BACKUP_DIR.

Returns the absolute path of the restored backup on success. Raises RuntimeError on any failure (caller should catch and st.error).

Source code in admin/services/backup.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def restore_backup(backup_filename: str) -> str:
    """Atomically restore a backup into the live DB.

    - Verifies the backup file's integrity before touching anything.
    - Uses sqlite3's backup API (WAL-aware) to overwrite the live DB.
    - Removes stale -wal / -shm sidecar files.
    - Refuses to run if the backup path escapes BACKUP_DIR.

    Returns the absolute path of the restored backup on success.
    Raises RuntimeError on any failure (caller should catch and st.error).
    """
    # ---- 1) Path safety: never trust caller-supplied filenames ----
    if not isinstance(backup_filename, str) or not backup_filename:
        raise RuntimeError("Invalid backup filename.")
    # Reject BOTH '/' and '\\' explicitly. os.path.basename() alone is
    # platform-dependent: on Linux '\\' is not a separator, so a name
    # like 'sub\\dir\\file.db' would slip through.
    if "/" in backup_filename or "\\" in backup_filename:
        raise RuntimeError("Backup filename must not contain path separators.")

    backup_path = os.path.join(BACKUP_DIR, backup_filename)
    if not os.path.isfile(backup_path):
        raise RuntimeError(f"Backup not found: {backup_filename}")
    if not os.path.isfile(DB_PATH):
        raise RuntimeError(f"Live DB not found: {DB_PATH}")

    # ---- 2) Verify the backup before we destroy the live DB ----
    try:
        chk_con = sqlite3.connect(backup_path)
        try:
            res = chk_con.execute("PRAGMA quick_check").fetchone()
        finally:
            chk_con.close()
    except sqlite3.DatabaseError as e:
        raise RuntimeError(f"Backup is not a valid SQLite DB: {e}") from e

    if not res or res[0] != "ok":
        raise RuntimeError(f"Backup integrity check failed: {res}")

    # ---- 3) Copy the backup INTO the live DB (atomic, WAL-aware) ----
    src = sqlite3.connect(backup_path)
    try:
        dst = sqlite3.connect(DB_PATH)
        try:
            dst.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            src.backup(dst)
            dst.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        finally:
            dst.close()
    finally:
        src.close()

    # ---- 4) Remove stale sidecar files left over from before ----
    for suffix in ("-wal", "-shm"):
        side = DB_PATH + suffix
        if os.path.exists(side):
            with contextlib.suppress(OSError):
                os.remove(side)

    return os.path.abspath(backup_path)

Audit - admin.services.audit

admin.services.audit.init_audit 

init_audit() -> None
Source code in admin/services/audit.py 
 9
10
11
12
def init_audit() -> None:
    db_exec("""CREATE TABLE IF NOT EXISTS audit_log(
        id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, user TEXT,
        action TEXT, entity TEXT, entity_id TEXT, details TEXT)""")

admin.services.audit.log_audit 

log_audit(action: str, entity: str, entity_id, details: str = '') -> None
Source code in admin/services/audit.py 
15
16
17
18
19
20
21
22
23
24
25
26
def log_audit(action: str, entity: str, entity_id, details: str = "") -> None:
    db_exec(
        "INSERT INTO audit_log(ts,user,action,entity,entity_id,details) " "VALUES(?,?,?,?,?,?)",
        (
            datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            current_user(),
            action,
            entity,
            str(entity_id),
            details,
        ),
    )

Anomaly detection - radar.services.anomaly

radar.services.anomaly.detect_anomalies 

detect_anomalies(cdrs_df: DataFrame, cells_df: DataFrame, sig_findings: dict) -> list[dict]

Return a list of anomaly dicts (type, entity, value, z_score, ...).

Detects
  • Hourly traffic outliers
  • City traffic outliers
  • Weak cells outliers
  • System-wide encryption failure spikes
  • CLIR abuse spikes
  • VoLTE fallback routing anomalies
Source code in radar/services/anomaly.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def detect_anomalies(
    cdrs_df: pd.DataFrame, cells_df: pd.DataFrame, sig_findings: dict
) -> list[dict]:
    """Return a list of anomaly dicts (type, entity, value, z_score, ...).

    Detects:
      - Hourly traffic outliers
      - City traffic outliers
      - Weak cells outliers
      - System-wide encryption failure spikes
      - CLIR abuse spikes
      - VoLTE fallback routing anomalies
    """
    anomalies: list[dict] = []
    if cdrs_df is None or not len(cdrs_df):
        return anomalies

    # 1. Hourly traffic anomalies
    hourly = cdrs_df.groupby("hour").size()
    if len(hourly) > 3:
        mean, std = hourly.mean(), hourly.std()
        if std > 0:
            for hour, val in hourly.items():
                z = (val - mean) / std
                if abs(z) > Z_MEDIUM:
                    anomalies.append(
                        {
                            "type": "Hourly Traffic",
                            "entity": f"Hour {hour:02d}:00",
                            "value": int(val),
                            "expected": int(mean),
                            "z_score": round(z, 2),
                            "severity": _severity(abs(z)),
                            "desc": f"Traffic {z:+.2f}σ from mean ({int(mean)})",
                        }
                    )

    # 2. City traffic anomalies
    city_counts = cdrs_df.groupby("city").size()
    if len(city_counts) > 3:
        mean, std = city_counts.mean(), city_counts.std()
        if std > 0:
            for city, val in city_counts.items():
                z = (val - mean) / std
                if abs(z) > Z_MEDIUM:
                    anomalies.append(
                        {
                            "type": "City Traffic",
                            "entity": city,
                            "value": int(val),
                            "expected": int(mean),
                            "z_score": round(z, 2),
                            "severity": _severity(abs(z)),
                            "desc": f"Traffic {z:+.2f}σ from mean",
                        }
                    )

    # 3. Weak cells
    weak = sig_findings.get("weak_cells", [])
    if weak:
        vals = np.array([c[1] for c in weak])
        if len(vals) > 1:
            mean, std = vals.mean(), vals.std()
            if std > 0:
                for cell_id, val in weak:
                    z = (val - mean) / std
                    if abs(z) > Z_MEDIUM:
                        anomalies.append(
                            {
                                "type": "Weak Cell",
                                "entity": cell_id,
                                "value": int(val),
                                "expected": int(mean),
                                "z_score": round(z, 2),
                                "severity": _severity(abs(z)),
                                "desc": f"Bad samples {z:+.2f}σ from mean",
                            }
                        )

    # 4. Encryption failures spike
    enc_fails = sig_findings.get("encryption_failures", [])
    if len(enc_fails) > 5:
        anomalies.append(
            {
                "type": "Encryption Failures",
                "entity": "System-wide",
                "value": len(enc_fails),
                "expected": 5,
                "z_score": round((len(enc_fails) - 5) / 3, 2),
                "severity": "HIGH" if len(enc_fails) > 20 else "MEDIUM",
                "desc": f"Unusually high failure count ({len(enc_fails)})",
            }
        )

    # 5. CLIR abuse spike
    clir_abuse = sig_findings.get("clir_abuse", [])
    if len(clir_abuse) > 20:
        anomalies.append(
            {
                "type": "CLIR Abuse Spike",
                "entity": "System-wide",
                "value": len(clir_abuse),
                "expected": 10,
                "z_score": round((len(clir_abuse) - 10) / 5, 2),
                "severity": "HIGH" if len(clir_abuse) > 40 else "MEDIUM",
                "desc": f"Unauthorized CLIR events ({len(clir_abuse)})",
            }
        )

    # 6. Voice bearer distribution anomaly
    if "voice_bearer" in cdrs_df.columns:
        v = cdrs_df[cdrs_df["call_type"] == "voice"]
        if len(v):
            dist = v["voice_bearer"].value_counts(normalize=True)
            tech_mix = cdrs_df["tech"].value_counts(normalize=True)
            modern = tech_mix.get("4G", 0) + tech_mix.get("5G", 0)
            csfb_pct = dist.get("CSFB", 0)
            if csfb_pct > 0.8 and modern > 0.5:
                anomalies.append(
                    {
                        "type": "VoLTE Fallback",
                        "entity": "Voice routing",
                        "value": f"{csfb_pct*100:.0f}%",
                        "expected": "20-40%",
                        "z_score": 2.5,
                        "severity": "MEDIUM",
                        "desc": (
                            f"High CSFB ({csfb_pct*100:.0f}%) despite "
                            f"{modern*100:.0f}% modern tech"
                        ),
                    }
                )

    return anomalies

Period comparison - radar.services.period

radar.services.period.split_periods 

split_periods(cdrs_f: DataFrame, split_ratio: float = 0.5)

Split a time-sorted CDR DataFrame into (previous, current).

Source code in radar/services/period.py 
15
16
17
18
19
20
21
def split_periods(cdrs_f: pd.DataFrame, split_ratio: float = 0.5):
    """Split a time-sorted CDR DataFrame into (previous, current)."""
    if cdrs_f is None or not len(cdrs_f):
        return None, None
    s = cdrs_f.sort_values("ts").reset_index(drop=True)
    idx = int(len(s) * split_ratio)
    return s.iloc[:idx], s.iloc[idx:]

radar.services.period.period_stats 

period_stats(df) -> dict

Return a dict of per-period totals. Empty for empty input.

Source code in radar/services/period.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def period_stats(df) -> dict:
    """Return a dict of per-period totals. Empty for empty input."""
    if df is None or not len(df):
        return {
            "cdr": 0,
            "bytes": 0,
            "voice": 0,
            "sms": 0,
            "mms": 0,
            "data": 0,
            "rcs": 0,
            "minutes": 0,
        }
    return {
        "cdr": len(df),
        "bytes": df["bytes"].sum(),
        "voice": int((df["call_type"] == "voice").sum()),
        "sms": int((df["call_type"] == "sms").sum()),
        "mms": int((df["call_type"] == "mms").sum()),
        "data": int((df["call_type"] == "data").sum()),
        "rcs": int((df["call_type"] == "rcs").sum()),
        "minutes": df.loc[df["call_type"] == "voice", "duration_sec"].sum() / 60,
    }

radar.services.period.delta_pct 

delta_pct(curr: float, prev: float)

Percent change from prev to curr. None when undefined.

Source code in radar/services/period.py 
49
50
51
52
53
def delta_pct(curr: float, prev: float):
    """Percent change from prev to curr. None when undefined."""
    if prev == 0:
        return None if curr == 0 else 100.0
    return (curr - prev) / prev * 100

PDF reports - radar.services.pdf

radar.services.pdf.build_pdf 

build_pdf(title: str, sections: list[dict]) -> bytes | None

Build a PDF report and return its bytes (or None if reportlab absent).

Each section is a dict with optional keys

heading : str text : str table : pandas.DataFrame

Source code in radar/services/pdf.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def build_pdf(title: str, sections: list[dict]) -> bytes | None:
    """Build a PDF report and return its bytes (or None if reportlab absent).

    Each section is a dict with optional keys:
        heading : str
        text    : str
        table   : pandas.DataFrame
    """
    if not HAS_PDF:
        return None
    try:
        buf = io.BytesIO()
        doc = SimpleDocTemplate(
            buf,
            pagesize=A4,
            leftMargin=0.6 * inch,
            rightMargin=0.6 * inch,
            topMargin=0.6 * inch,
            bottomMargin=0.6 * inch,
        )
        styles = getSampleStyleSheet()
        title_style = ParagraphStyle(
            "Title",
            parent=styles["Title"],
            textColor=rl_colors.HexColor("#0f172a"),
            fontSize=20,
            spaceAfter=8,
        )
        h2_style = ParagraphStyle(
            "H2",
            parent=styles["Heading2"],
            textColor=rl_colors.HexColor("#f6821f"),
            fontSize=14,
            spaceBefore=12,
            spaceAfter=6,
        )
        body_style = styles["BodyText"]
        body_style.fontSize = 10
        body_style.textColor = rl_colors.HexColor("#334155")

        story = []
        story.append(Paragraph(title, title_style))
        story.append(
            Paragraph(
                f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | "
                f"Operator: TELECOM | Mode: LOCAL-ONLY",
                body_style,
            )
        )
        story.append(Spacer(1, 12))

        for s in sections:
            if s.get("heading"):
                story.append(Paragraph(s["heading"], h2_style))
            if s.get("text"):
                story.append(Paragraph(s["text"], body_style))
                story.append(Spacer(1, 6))
            if s.get("table") is not None and len(s["table"]) > 0:
                df = s["table"].head(20)
                data = [list(df.columns)] + df.astype(str).values.tolist()
                t = Table(data, hAlign="LEFT")
                t.setStyle(
                    TableStyle(
                        [
                            ("BACKGROUND", (0, 0), (-1, 0), rl_colors.HexColor("#f6821f")),
                            ("TEXTCOLOR", (0, 0), (-1, 0), rl_colors.white),
                            ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
                            ("FONTSIZE", (0, 0), (-1, -1), 8),
                            ("GRID", (0, 0), (-1, -1), 0.25, rl_colors.grey),
                            (
                                "ROWBACKGROUNDS",
                                (0, 1),
                                (-1, -1),
                                [rl_colors.white, rl_colors.HexColor("#f8fafc")],
                            ),
                            ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
                            ("PADDING", (0, 0), (-1, -1), 4),
                        ]
                    )
                )
                story.append(t)
                story.append(Spacer(1, 12))

        doc.build(story)
        return buf.getvalue()
    except Exception:
        return None