Play Policy Insights

scripts/generate_report.py

#!/usr/bin/env python3

# Copyright 2026 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Generates a comprehensive compliance report from synthesized findings."""

import argparse
from datetime import datetime
import glob
import json
import os
import re
import sys
import play_store_scraper
from template_engine import render_template

# Precompiled regex patterns at module scope for performance optimizations
EMOJI_PATTERN = re.compile(
    r"[\U00010000-\U0010ffff\u2600-\u27bf\u2300-\u23ff\u2b50]+",
    flags=re.UNICODE,
)
CODE_LINK_PATTERN = re.compile(r"^(.*?)(?:[:\s]+L?(\d+))?$")
SEVERITY_CLEAN_PATTERN = re.compile(r"['\"`*🔴🟡🔵\s]+")


def run_scraper(package_name, output_dir):
  """Invokes the Play Store scraper directly."""
  try:
    print(f"Auditing {package_name} from Play Store...", file=sys.stderr)
    result = play_store_scraper.scrape_app_details(package_name)

    # Save to temp for consistency with existing report logic if needed
    os.makedirs(output_dir, exist_ok=True)
    out_path = os.path.join(output_dir, "play_store_declaration.json")
    with open(out_path, "w", encoding="utf-8") as f:
      json.dump(result, f, indent=4, sort_keys=True)

    return result
  except Exception as e:
    print(f"Warning: Play Store scrape failed: {e}", file=sys.stderr)
  return {}


def strip_emojis(text):
  """Removes emoji characters from a string."""
  if not isinstance(text, str):
    return text
  return EMOJI_PATTERN.sub(r"", text)


def format_code_link(file_val):
  """Formats file paths and line numbers into Markdown links."""
  if not file_val or file_val == "N/A":
    return "N/A"
  files = []
  if isinstance(file_val, str):
    if "," in file_val:
      files = [f.strip() for f in file_val.split(",")]
    else:
      files = [file_val.strip()]
  elif isinstance(file_val, list):
    files = [str(f).strip() for f in file_val]
  formatted_links = []
  for f in files:
    f_str = str(f)
    if f_str.startswith("["):
      formatted_links.append(f_str)
      continue
    match = CODE_LINK_PATTERN.search(f_str)
    if match:
      path = match.group(1)
      line = match.group(2)
      basename = os.path.basename(path)
      if line:
        link = f"[{basename}:L{line}](file://{path}#L{line})"
      else:
        link = f"[{basename}](file://{path})"
      formatted_links.append(link)
    else:
      formatted_links.append(f_str)
  return ", ".join(formatted_links)


def render_table(
    items, columns, column_names, empty_message="No items identified."
):
  """Renders a Markdown table from a list of dictionaries."""
  if not items:
    return f"\n* {empty_message}\n"

  widths = [len(name) for name in column_names]
  rows = []
  for item in items:
    row = []
    for col in columns:
      val = str(item.get(col, "N/A")).replace("|", "\\|").replace("\n", "<br>")
      row.append(val)
    rows.append(row)
    for i, val in enumerate(row):
      widths[i] = max(widths[i], len(val))

  def format_row(vals):
    return (
        "| "
        + " | ".join(val.ljust(widths[i]) for i, val in enumerate(vals))
        + " |"
    )

  header = format_row(column_names)
  separator = (
      "| " + " | ".join("-" * widths[i] for i in range(len(widths))) + " |"
  )
  body = "\n".join(format_row(row) for row in rows)

  return f"\n{header}\n{separator}\n{body}\n"


def parse_boolean(val):
  """Safely converts boolean or sloppy string yes/no/true/false."""
  if isinstance(val, bool):
    return val
  if isinstance(val, str):
    v_clean = val.strip().lower()
    return v_clean in ("yes", "true", "1")
  return False


def parse_disclosure_status(val):
  """Safely parses disclosure enums or visual yes/no."""
  if not val:
    return "No"

  val_clean = str(val).strip().upper()
  if "EXEMPT" in val_clean:
    return "Exempt (Obvious)"
  if "DISCLOSED" in val_clean or "YES" in val_clean or "TRUE" in val_clean:
    return "Yes"

  return "No"


def clean_severity(sev_str):
  """Robustly cleans severity strings by removing quotes, asterisks,
  emojis, and whitespace.
  """
  if not isinstance(sev_str, str):
    return "SUGGESTION"

  # Convert to uppercase and strip formatting characters
  cleaned = sev_str.upper()
  cleaned = SEVERITY_CLEAN_PATTERN.sub("", cleaned)

  if "CRITICAL" in cleaned:
    return "CRITICAL"
  if "IMPORTANT" in cleaned:
    return "IMPORTANT"
  if "SUGGESTION" in cleaned:
    return "SUGGESTION"

  return "SUGGESTION"


def render_finding_card(finding, policy_refs=None):
  """Renders a detailed finding card in Markdown."""
  severity = clean_severity(finding.get("severity"))
  severity_map = {
      "CRITICAL": "🔴 Critical",
      "IMPORTANT": "🟡 Important",
      "SUGGESTION": "🔵 Suggestion",
  }
  visual_severity = severity_map.get(severity, "🔵 Suggestion")

  title = strip_emojis(
      finding.get("issue") or finding.get("issue_summary") or "Policy Risk"
  )
  p_id = (
      finding.get("policy_id")
      or finding.get("policy_reference")
      or "Unknown Policy"
  )
  policy_link = f"**{p_id}**"

  policy_info = policy_refs.get(p_id) if policy_refs else None

  policy_link = f"**{p_id}**"
  policy_urls = []
  if p_id and policy_info:
    policy_link = f"**{policy_info.get('name', p_id)}**"
    policy_urls = policy_info.get("urls", [])

  files = finding.get("files_involved")
  has_local_override = False
  local_override_file = None

  if files and isinstance(files, list):
    for f in files:
      if "play_store_declaration.json" in f:
        has_local_override = True
        local_override_file = f
        break

  files_formatted = format_code_link(files)
  recommendation = finding.get("recommendation", "Review and remediate.")

  if has_local_override:
    recommendation = (
        "Update the local Play Store declaration file at "
        f"{format_code_link(local_override_file)}."
    )

  card = f"\n#### {title}\n"
  card += f"- **Policy**: {policy_link}\n"
  card += f"- **Severity**: {visual_severity}\n"
  card += f"- **Files**: {files_formatted}\n"
  card += f"- **Evidence**: {finding.get('evidence', 'N/A')}\n"
  card += f"- **Recommendation**: {recommendation}\n"

  if policy_urls:
    card += "- **References**:\n"
    for url in policy_urls:
      card += f"  - {url}\n"

  return card


def load_json(file_path):
  """Loads JSON from a file if it exists."""
  if os.path.exists(file_path):
    try:
      with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
        return json.load(f)
    except Exception as e:
      print(f"Warning: Failed to load {file_path}: {e}", file=sys.stderr)
  return None


def aggregate_findings(temp_dir, taxonomy):
  """Consolidates audit findings from workers.

  Produces a unified report data object.
  """
  play_store_info = (
      load_json(os.path.join(temp_dir, "play_store_info.json")) or {}
  )
  manifest_details = (
      load_json(os.path.join(temp_dir, "manifest_details.json")) or {}
  )

  aggregated_findings_path = os.path.join(temp_dir, "aggregated_findings.json")

  raw_findings = []
  critic_decisions = {}

  if os.path.exists(aggregated_findings_path):
    master_data = load_json(aggregated_findings_path) or {}
    raw_findings = master_data.get("findings", [])

    # Load all chunked critic outputs: critic_output_*.json
    critic_files = sorted(
        glob.glob(os.path.join(temp_dir, "critic_output_*.json"))
    )
    for c_file in critic_files:
      c_data = load_json(c_file) or {}
      for fid, dec in c_data.items():
        critic_decisions[fid] = dec

  # 2. Process Findings and Apply Decoupled Critic Verdicts
  identified_risks = []
  manual_review_needed = []
  data_safety_inventory = []

  # Sort raw findings by finding_id numerically for stable processing order
  def finding_sort_key(f):
    fid = f.get("finding_id", "0")
    try:
      return int(fid)
    except (ValueError, TypeError):
      return 0

  sorted_raw_findings = sorted(raw_findings, key=finding_sort_key)

  for finding in sorted_raw_findings:
    p_id = finding.get("policy_id", "Unknown")
    psl = finding.get("psl_constant")
    fid = finding.get("finding_id")
    wf = finding.get("worker_file")

    # Retrieve decision
    decision = critic_decisions.get(str(fid))
    if not decision:
      # Automatically approve/verify findings that bypassed the Critic
      # (like SUGGESTIONs)
      decision = {
          "action": "VERIFIED",
          "confidence": "High",
          "critic_justification": (
              "Automatically verified (bypassed critic review)."
          ),
      }

    action = str(decision.get("action", "VERIFIED")).upper().strip()
    confidence = decision.get("confidence", "Medium")
    justification = decision.get("critic_justification", "Verified by Critic.")

    # Apply Critic's editorial overrides if specified
    severity = decision.get("severity") or finding.get("severity", "SUGGESTION")
    severity = clean_severity(severity)

    issue_summary = decision.get("issue_summary") or finding.get(
        "issue_summary", "Unknown Issue"
    )
    recommendation = decision.get("recommendation") or finding.get(
        "recommendation", ""
    )

    files_involved = finding.get("files_involved", [])
    if isinstance(files_involved, list):
      files_involved = sorted([str(f) for f in files_involved])

    evidence = finding.get("evidence", "")

    reconstructed_finding = {
        "policy_id": p_id,
        "finding_id": fid or f"{wf}_{p_id}",
        "issue_summary": issue_summary,
        "severity": severity,
        "files_involved": files_involved,
        "evidence": (
            f"{evidence}\n\n**Critic Verification**: {justification}"
            f" (Confidence: {confidence})"
        ),
        "recommendation": recommendation,
    }

    # Programmatically evaluate if this is a compliant Data Safety finding
    # to avoid card clutter
    is_compliant_ds = False
    if "is_transferred" in finding:
      is_transferred = parse_boolean(finding.get("is_transferred"))
      user_initiated = parse_boolean(finding.get("user_initiated"))
      is_third_party = parse_boolean(finding.get("is_third_party"))
      disc_status = (
          str(finding.get("prominent_disclosure_status", "MISSING"))
          .upper()
          .strip()
      )

      # Compliant if either local-only OR transmitted but disclosed/exempt
      # OR user initiated
      if not is_transferred:
        is_compliant_ds = True
      elif user_initiated:
        is_compliant_ds = True
      elif "DISCLOSED" in disc_status or "EXEMPT" in disc_status:
        is_compliant_ds = True

    # Route the finding based on Critic's action
    if action == "VERIFIED":
      if not is_compliant_ds:
        identified_risks.append(reconstructed_finding)
    elif action == "MANUAL_REVIEW":
      if not is_compliant_ds:
        manual_review_needed.append(reconstructed_finding)

    # 3. Process Data Safety Inventory Programmatically (If it's a DS Finding)
    if "is_transferred" in finding:
      # Create a localized copy of the data safety keys (psl_constant is
      # natively preserved)
      inv_item = {
          "psl_constant": psl,
          "is_transferred": parse_boolean(finding.get("is_transferred")),
          "user_initiated": parse_boolean(finding.get("user_initiated")),
          "is_third_party": parse_boolean(finding.get("is_third_party")),
          "prominent_disclosure_status": (
              str(finding.get("prominent_disclosure_status", "MISSING"))
              .upper()
              .strip()
          ),
          "purpose": finding.get("purpose", "N/A"),
          "linked_to_user": parse_boolean(finding.get("linked_to_user")),
          "behavioral_proof": evidence,
          "disclosure_proof": recommendation,
      }

      # Programmatically align/vet inventory based on Critic's verdict
      if action == "PRUNED":
        inv_item["is_transferred"] = False
        inv_item["purpose"] = "Local functionality only"
        inv_item["behavioral_proof"] = f"Pruned by Critic: {justification}"

      data_safety_inventory.append(inv_item)

  # 4. Consolidate and Deduplicate Data Safety Inventory by psl_constant
  merged_inventory = {}
  for item in data_safety_inventory:
    psl_id = item.get("psl_constant")
    if not psl_id:
      continue

    if psl_id not in merged_inventory:
      merged_inventory[psl_id] = {
          "psl_constant": psl_id,
          "is_transferred": False,
          "user_initiated": True,  # Default to True for AND logic
          "is_third_party": False,
          "linked_to_user": False,
          "prominent_disclosure_status": "EXEMPT",
          "purposes": set(),
          "behavioral_proofs": set(),
          "disclosure_proofs": set(),
      }

    current = merged_inventory[psl_id]

    # Transmission (True if any is True)
    if item.get("is_transferred"):
      current["is_transferred"] = True

    # User Initiated (True only if ALL are True)
    if not item.get("user_initiated"):
      current["user_initiated"] = False

    # Third Party (True if any is True)
    if item.get("is_third_party"):
      current["is_third_party"] = True

    # Linked (True if any is True)
    if item.get("linked_to_user"):
      current["linked_to_user"] = True

    # Disclosure logic precedence: MISSING > DISCLOSED > EXEMPT
    new_disc = (
        str(item.get("prominent_disclosure_status", "EXEMPT")).upper().strip()
    )
    curr_disc = current["prominent_disclosure_status"]

    if "MISSING" in curr_disc or "MISSING" in new_disc:
      current["prominent_disclosure_status"] = "MISSING"
    elif "DISCLOSED" in curr_disc or "DISCLOSED" in new_disc:
      current["prominent_disclosure_status"] = "DISCLOSED"
    else:
      current["prominent_disclosure_status"] = "EXEMPT"

    if item.get("purpose") and item.get("purpose") != "N/A":
      current["purposes"].add(str(item["purpose"]).strip())
    if item.get("behavioral_proof") and item.get("behavioral_proof") != "N/A":
      current["behavioral_proofs"].add(str(item["behavioral_proof"]).strip())
    if item.get("disclosure_proof") and item.get("disclosure_proof") != "N/A":
      current["disclosure_proofs"].add(str(item["disclosure_proof"]).strip())

  # Reconstruct the list with joined sets
  normalized_inventory = []
  sorted_psl_ids = sorted(merged_inventory.keys())
  for psl_id in sorted_psl_ids:
    item = merged_inventory[psl_id]
    reconstructed = {
        "psl_constant": psl_id,
        "is_transferred": "Yes" if item["is_transferred"] else "No",
        "user_initiated": "Yes" if item.get("user_initiated") else "No",
        "is_third_party": "Yes" if item.get("is_third_party") else "No",
        "linked_to_user": "Yes" if item["linked_to_user"] else "No",
        "prominent_disclosure_status": parse_disclosure_status(
            item["prominent_disclosure_status"]
        ),
        "purpose": ", ".join(sorted(list(item["purposes"]))) or "N/A",
        "behavioral_proof": (
            ", ".join(sorted(list(item["behavioral_proofs"]))) or "N/A"
        ),
        "disclosure_proof": (
            ", ".join(sorted(list(item["disclosure_proofs"]))) or "N/A"
        ),
    }
    normalized_inventory.append(reconstructed)

  # 5. Decorate and Filter Data Safety Inventory
  decorated_inventory = []
  local_access_only = []

  for item in normalized_inventory:
    psl_id = item.get("psl_constant")
    tax_info = taxonomy.get(psl_id, {"category": "Other", "data_type": "Other"})
    item["category"] = tax_info["category"]
    item["data_type"] = tax_info["data_type"]

    is_local = (
        item.get("is_transferred") == "No"
        or item.get("purpose") == "Local functionality only"
    )
    if is_local:
      local_access_only.append(item)
    else:
      decorated_inventory.append(item)

  # Cross-reference with Play Store
  matches = []
  mismatches = []
  is_published = play_store_info.get("is_published", False)

  if is_published:
    play_declarations = play_store_info.get("data_safety", {}).get(
        "data_collected", []
    )
    play_data_types = set()
    for category_dict in play_declarations:
      for type_dict in category_dict.get("types", []):
        play_data_types.add(type_dict.get("type"))

    for item in decorated_inventory:
      dt = item["data_type"]
      if dt in play_data_types:
        matches.append({"data_type": dt, "status": "Declared and detected"})
      else:
        is_obvious = "Exempt" in item.get(
            "prominent_disclosure_status", ""
        ) or "Obvious" in item.get("prominent_disclosure_status", "")
        mismatches.append({
            "data_type": dt,
            "local_view": (
                f"Detected in code (Evidence: {item.get('behavioral_proof')})"
            ),
            "play_view": "Not declared in Play Store",
            "status": (
                "Exempt: Obvious core functionality"
                if is_obvious
                else "Discrepancy"
            ),
        })

    # Check for unjustified declarations (in Play Store but not in code)
    detected_data_types = {
        item["data_type"] for item in decorated_inventory + local_access_only
    }
    for category_dict in play_declarations:
      for type_dict in category_dict.get("types", []):
        dt = type_dict.get("type")
        if dt not in detected_data_types:
          mismatches.append({
              "data_type": dt,
              "local_view": "Not detected in code",
              "play_view": "Declared in Play Store",
              "status": "Discrepancy",
          })

  # 7. Determine Compliance
  overall_compliance = "Compliant"
  critical_risks = any(
      clean_severity(r.get("severity")) == "CRITICAL" for r in identified_risks
  )
  active_mismatches = any(m.get("status") == "Discrepancy" for m in mismatches)

  if critical_risks or active_mismatches:
    overall_compliance = "Non-compliant"
  elif manual_review_needed or any(
      clean_severity(r.get("severity")) == "IMPORTANT" for r in identified_risks
  ):
    overall_compliance = "Needs review"

  # 8. Sort result lists for deterministic output
  def report_finding_sort_key(f):
    pid = f.get("policy_id", "")
    fid = f.get("finding_id", "0")
    try:
      num_fid = int(fid)
    except (ValueError, TypeError):
      num_fid = 0
    return (pid, num_fid)

  identified_risks.sort(key=report_finding_sort_key)
  manual_review_needed.sort(key=lambda x: x.get("issue_summary", ""))
  matches.sort(key=lambda x: x.get("data_type", ""))
  mismatches.sort(key=lambda x: x.get("data_type", ""))

  # 9. Generate Summary

  risk_count = len(identified_risks)
  mismatch_count = len(
      [m for m in mismatches if m.get("status") == "Discrepancy"]
  )
  exempt_count = len([
      m
      for m in mismatches
      if m.get("status") == "Exempt: Obvious core functionality"
  ])

  summary_parts = []
  summary_parts.append(
      "The automated audit of"
      f" {play_store_info.get('title', 'the application')} is complete."
  )
  summary_parts.append(
      f"Identified {risk_count} potential policy risks and {mismatch_count}"
      " active Data Safety discrepancies."
  )
  if exempt_count > 0:
    summary_parts.append(
        f"{exempt_count} detections were flagged as 'Obvious' and exempt from"
        " prominent disclosure."
    )

  if overall_compliance == "Non-compliant":
    summary_parts.append(
        "Immediate remediation is required for critical findings and"
        " declaration mismatches."
    )
  elif overall_compliance == "Needs review":
    summary_parts.append(
        "Manual review is recommended for several ambiguous findings."
    )
  else:
    summary_parts.append(
        "The application appears broadly compliant with analyzed policies."
    )

  # 9. Final Report Construction
  pkg_name = (
      play_store_info.get("package_name")
      or manifest_details.get("package_name")
  )
  if not pkg_name or pkg_name == "unknown":
    pkg_name = "unknown"

  report = {
      "overall_compliance": overall_compliance,
      "summary": " ".join(summary_parts),
      "package_name": pkg_name,
      "is_published": is_published,
      "identified_risks": identified_risks,
      "data_safety_comparison": {"matches": matches, "mismatches": mismatches},
      "local_data_access": [
          {
              "data_type": item["data_type"],
              "category": item["category"],
              "evidence": item.get("behavioral_proof"),
          }
          for item in local_access_only
      ],
      "manual_review_needed": [
          {"issue_summary": r.get("issue_summary")}
          for r in manual_review_needed
      ],
      "suggested_data_safety_declaration": {
          "collected_data": [
              {
                  "data_type": item["data_type"],
                  "purpose": item.get("purpose"),
                  "linked_to_user": item.get("linked_to_user") == "Yes",
              }
              for item in decorated_inventory
          ]
      },
  }

  return report


def main():
  """Main entry point for report generation."""
  parser = argparse.ArgumentParser(description="Compliance report generator.")
  parser.add_argument(
      "temp_dir",
      help="Path to the temporary scratch directory containing audit findings.",
  )

  args = parser.parse_args()
  temp_dir = os.path.abspath(args.temp_dir)
  repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

  policy_refs = (
      load_json(os.path.join(repo_root, "resources", "policies.json"))
  )
  if not policy_refs:
    policy_refs = {}
  taxonomy = policy_refs.get("data_safety_section", {}).get("taxonomy", {})

  manifest_details = (
      load_json(os.path.join(temp_dir, "manifest_details.json")) or {}
  )
  package_name = manifest_details.get("package_name")

  # Load Play Store info (potentially ingested by orchestrator)
  play_store_info_path = os.path.join(temp_dir, "play_store_info.json")
  play_store_info = {}
  if os.path.exists(play_store_info_path):
    play_store_info = load_json(play_store_info_path) or {}

  if not play_store_info and package_name:
    play_store_info = run_scraper(package_name, temp_dir)
    # Cache it for report reuse
    with open(play_store_info_path, "w", encoding="utf-8") as f:
      json.dump(play_store_info, f, indent=4, sort_keys=True)

  report_data = aggregate_findings(temp_dir, taxonomy)

  template_path = os.path.join(
      repo_root, "resources", "compliance_report_template.md"
  )
  if not os.path.exists(template_path):
    print(f"Error: Template not found at {template_path}", file=sys.stderr)
    sys.exit(1)

  try:
    with open(template_path, "r", encoding="utf-8") as f:
      content = f.read()
  except Exception as e:
    print(f"Error reading template: {e}", file=sys.stderr)
    sys.exit(1)

  output_path = os.path.join(temp_dir, "compliance_report.md")

  violations = report_data.get("identified_risks", [])
  mismatches = report_data.get("data_safety_comparison", {}).get(
      "mismatches", []
  )
  manual = report_data.get("manual_review_needed", [])

  compliance_raw = report_data.get("overall_compliance") or "Needs review"
  status_map = {
      "Non-compliant": "🔴 Non-Compliant",
      "Needs review": "🟡 Needs Review",
      "Compliant": "🟢 Compliant",
  }
  compliance_ui = status_map.get(compliance_raw, "🟡 Needs Review")

  app_id = report_data.get("package_name")
  if not app_id or app_id == "unknown":
    app_id = (
        play_store_info.get("package_name")
        or manifest_details.get("package_name")
        or "Unknown App"
    )

  findings_content = ""
  if not violations:
    findings_content = "\n* No policy risks identified in this scan.\n"
  else:
    grouped_findings = {}
    for v in violations:
      p_id = v.get("policy_id")
      macro_cat = "Other Policies"
      if p_id and p_id in policy_refs:
        macro_cat = policy_refs[p_id].get("category", "Other Policies")

      if macro_cat not in grouped_findings:
        grouped_findings[macro_cat] = []
      grouped_findings[macro_cat].append(v)

    ordered_cats = [
        "Restricted Content",
        "Privacy, Deception and Device Abuse",
        "Monetization and Ads",
        "Store Listing and Promotion",
        "Developer Account Management",
    ]

    all_cats = ordered_cats + [
        cat for cat in grouped_findings if cat not in ordered_cats
    ]

    for cat in all_cats:
      if cat in grouped_findings:
        findings_content += f"\n### {cat}\n"
        for v in grouped_findings[cat]:
          findings_content += render_finding_card(v, policy_refs) + "\n"

  is_not_published = report_data.get("is_published") is False

  d_table_msg = "Code detection matches Play Store declarations."
  if is_not_published:
    d_table_msg = (
        "N/A - App is not yet published. No declarations found for comparison."
    )

  d_table = render_table(
      mismatches,
      ["data_type", "local_view", "play_view", "status"],
      ["Data Type", "Code Detection", "Play Store Declaration", "Status"],
      d_table_msg,
  )

  # Local Access Section
  local_access = report_data.get("local_data_access", [])
  la_content = ""
  if local_access:
    la_table = render_table(
        local_access,
        ["data_type", "category", "evidence"],
        ["Data Type", "Category", "Access Evidence"],
        "No local-only access detected.",
    )
    la_content = (
        "## Local Data Access (No Transmission)\nThe following data types are"
        " accessed by the code but no evidence of network transmission or"
        " exfiltration was detected. These typically do not require a"
        f" 'Collection' declaration in the Data Safety section.\n{la_table}"
    )

  # Suggested Declaration (Conditional Section)
  suggested_dec = report_data.get("suggested_data_safety_declaration", {})
  dec_content = ""
  if suggested_dec and any(suggested_dec.values()):
    dec_content = (
        "### Suggested Data Safety Declaration"
        f" Updates\n```json\n{json.dumps(suggested_dec, indent=4)}\n```"
    )

  checklist_items = []
  for m in manual:
    m_issue = strip_emojis(
        m.get("issue") or m.get("issue_summary") or "Review item"
    )
    checklist_items.append(f"- [ ] {m_issue}")

  risk_categories = set(
      [v.get("category") for v in violations if v.get("category")]
  )
  mapping = {
      "Undeclared Collection": "Update Data Safety section in Play Console.",
      "Permissions": "Review and minimize requested permissions.",
  }
  for cat in risk_categories:
    if cat in mapping:
      checklist_items.append(f"- [ ] {mapping[cat]}")
  if not checklist_items:
    checklist_items.append("- [ ] Review all identified policy risks.")

  template_context = {
      "overall_compliance": compliance_ui,
      "current_date": datetime.now().strftime("%Y-%m-%d"),
      "app_name_id": app_id,
      "findings_detail": findings_content,
      "data_safety_table": d_table,
      "local_access_section": la_content,
      "suggested_declaration_section": dec_content,
      "personalized_checklist": "\n".join(checklist_items),
  }

  content = render_template(content, template_context)

  with open(output_path, "w", encoding="utf-8") as f:
    f.write(content)

  json_output_path = output_path.replace(".md", ".json")
  with open(json_output_path, "w", encoding="utf-8") as f:
    json.dump(
        {
            "metadata": {
                "app_id": app_id,
                "scan_date": datetime.now().isoformat(),
                "overall_compliance": compliance_raw,
            },
            "findings": violations,
            "data_safety_mismatches": mismatches,
            "manual_review": manual,
            "suggested_declaration": suggested_dec,
        },
        f,
        indent=4,
        sort_keys=True,
    )
  print(f"Reports generated successfully at {output_path}")


if __name__ == "__main__":
  main()