#!/usr/bin/env bash set -euo pipefail cat > /tmp/export_sonar.py << 'PYEOF' import base64 import csv import json import os import sys import time import urllib.parse import urllib.request TOKEN = os.getenv("SONAR_TOKEN", "squ_2b30e2fa92144d9fe47c238b95ce7063b4ab7701") HOST = os.getenv("SONAR_HOST", "https://sonar.taaazzz-prog.fr").rstrip("/") PROJECT = os.getenv("SONAR_PROJECT", "faildaily") BRANCH = os.getenv("SONAR_BRANCH", "").strip() PAGE_SIZE = int(os.getenv("SONAR_PAGE_SIZE", "100")) ISSUES_CSV = "sonar-issues.csv" ISSUES_DETAILED_CSV = "sonar-issues-detailed.csv" HOTSPOTS_CSV = "sonar-hotspots.csv" HOTSPOTS_DETAILED_CSV = "sonar-hotspots-detailed.csv" COVERAGE_CSV = "sonar-coverage.csv" COVERAGE_DETAILED_CSV = "sonar-coverage-detailed.csv" DUPLICATION_CSV = "sonar-duplications.csv" DUPLICATION_DETAILED_CSV = "sonar-duplications-detailed.csv" SUMMARY_JSON = "sonar-summary.json" ISSUE_FIELDS = [ "severity", "type", "file", "line", "message", ] ISSUE_DETAILED_FIELDS = [ "key", "severity", "type", "status", "resolution", "rule", "project", "component", "file", "line", "hash", "effort", "debt", "author", "assignee", "creation_date", "update_date", "clean_code_attribute", "clean_code_attribute_category", "impacts", "tags", "text_range", "message", ] HOTSPOT_FIELDS = [ "status", "vulnerability_probability", "security_category", "file", "line", "rule_key", "message", ] HOTSPOT_DETAILED_FIELDS = [ "key", "project", "component", "file", "line", "status", "security_category", "vulnerability_probability", "rule_key", "author", "creation_date", "update_date", "text_range", "message", ] COVERAGE_FIELDS = [ "file", "coverage", "uncovered_lines", "uncovered_conditions", "lines_to_cover", ] COVERAGE_DETAILED_FIELDS = [ "file", "name", "qualifier", "language", "coverage", "line_coverage", "branch_coverage", "lines_to_cover", "uncovered_lines", "conditions_to_cover", "uncovered_conditions", "new_coverage", "new_line_coverage", "new_branch_coverage", "new_lines_to_cover", "new_uncovered_lines", "new_conditions_to_cover", "new_uncovered_conditions", "ncloc", "complexity", "cognitive_complexity", "coverage_line_hist_data", "branch_coverage_hits_data", "conditions_by_line", "covered_conditions_by_line", ] DUPLICATION_FIELDS = [ "file", "duplicated_lines_density", "duplicated_lines", "duplicated_blocks", "new_duplicated_lines", "new_duplicated_blocks", "new_duplicated_lines_density", "ncloc", ] DUPLICATION_DETAILED_FIELDS = [ "file", "name", "qualifier", "language", "duplicated_lines_density", "duplicated_lines", "duplicated_blocks", "new_duplicated_lines", "new_duplicated_blocks", "new_duplicated_lines_density", "ncloc", ] PROJECT_METRICS = [ "coverage", "line_coverage", "branch_coverage", "lines_to_cover", "uncovered_lines", "conditions_to_cover", "uncovered_conditions", "new_coverage", "new_line_coverage", "new_branch_coverage", "new_lines_to_cover", "new_uncovered_lines", "new_conditions_to_cover", "new_uncovered_conditions", "violations", "bugs", "vulnerabilities", "code_smells", "duplicated_lines_density", "duplicated_lines", "duplicated_blocks", "new_duplicated_lines", "new_duplicated_blocks", "new_duplicated_lines_density", "ncloc", "complexity", "cognitive_complexity", ] OPTIONAL_FILE_METRICS = [ "coverage_line_hist_data", "branch_coverage_hits_data", "conditions_by_line", "covered_conditions_by_line", ] def build_url(path, params=None): params = dict(params or {}) if BRANCH: params.setdefault("branch", BRANCH) query = urllib.parse.urlencode(params) return f"{HOST}{path}" + (f"?{query}" if query else "") def fetch_json(path, params=None, retries=3, backoff=1.0): url = build_url(path, params) req = urllib.request.Request(url) creds = base64.b64encode((TOKEN + ":").encode()).decode() req.add_header("Authorization", "Basic " + creds) req.add_header("Accept", "application/json") for attempt in range(1, retries + 1): try: with urllib.request.urlopen(req) as response: return json.load(response) except Exception as exc: if attempt == retries: raise RuntimeError(f"Échec Sonar API après {retries} tentatives: {url}") from exc time.sleep(backoff * attempt) def fetch_json_once(path, params=None): url = build_url(path, params) req = urllib.request.Request(url) creds = base64.b64encode((TOKEN + ":").encode()).decode() req.add_header("Authorization", "Basic " + creds) req.add_header("Accept", "application/json") with urllib.request.urlopen(req) as response: return json.load(response) def sanitize_text(value): if value is None: return "" return str(value).replace("\n", " ").replace("\r", " ").strip() def split_component_key(value): if not value: return "", "" if ":" in value: return value.split(":", 1) return "", value def component_path(component): path = component.get("path") if path: return path component_value = component.get("component") or component.get("key", "") _, resolved_path = split_component_key(component_value) return resolved_path def measure_map(component): return {measure["metric"]: measure.get("value", "") for measure in component.get("measures", [])} def flatten_text_range(issue): text_range = issue.get("textRange") or {} return ":".join( [ str(text_range.get("startLine", "")), str(text_range.get("endLine", "")), str(text_range.get("startOffset", "")), str(text_range.get("endOffset", "")), ] ).strip(":") def flatten_impacts(issue): impacts = issue.get("impacts") or [] if not impacts: return "" return " | ".join( f"{impact.get('softwareQuality', '')}:{impact.get('severity', '')}" for impact in impacts ) def paginate_issues(): page = 1 total = 1 exported = 0 while exported < total: data = fetch_json( "/api/issues/search", { "projects": PROJECT, "statuses": "OPEN,CONFIRMED,REOPENED", "additionalFields": "_all", "ps": PAGE_SIZE, "p": page, }, ) issues = data.get("issues", []) total = data.get("total", 0) exported += len(issues) print(f"Issues page {page} — {exported}/{total}") yield issues page += 1 def paginate_hotspots(): page = 1 total = 1 exported = 0 while exported < total: data = fetch_json( "/api/hotspots/search", { "projectKey": PROJECT, "ps": PAGE_SIZE, "p": page, }, ) hotspots = data.get("hotspots", []) total = data.get("paging", {}).get("total", 0) exported += len(hotspots) print(f"Hotspots page {page} — {exported}/{total}") yield hotspots page += 1 def paginate_files(metric_keys, label): page = 1 total = 1 exported = 0 while exported < total: data = fetch_json( "/api/measures/component_tree", { "component": PROJECT, "metricKeys": ",".join(metric_keys), "qualifiers": "FIL", "ps": PAGE_SIZE, "p": page, }, ) components = data.get("components", []) total = data.get("paging", {}).get("total", 0) exported += len(components) print(f"{label} page {page} — {exported}/{total}") yield components page += 1 def export_issues(): with open(ISSUES_CSV, "w", newline="", encoding="utf-8") as simple_file, open( ISSUES_DETAILED_CSV, "w", newline="", encoding="utf-8" ) as detailed_file: simple_writer = csv.DictWriter(simple_file, fieldnames=ISSUE_FIELDS) detailed_writer = csv.DictWriter(detailed_file, fieldnames=ISSUE_DETAILED_FIELDS) simple_writer.writeheader() detailed_writer.writeheader() for issues in paginate_issues(): for issue in issues: file_path = component_path(issue) base_row = { "severity": issue.get("severity", ""), "type": issue.get("type", ""), "file": file_path, "line": issue.get("line", ""), "message": sanitize_text(issue.get("message", "")), } simple_writer.writerow(base_row) detailed_writer.writerow( { "key": issue.get("key", ""), "severity": issue.get("severity", ""), "type": issue.get("type", ""), "status": issue.get("status", ""), "resolution": issue.get("resolution", ""), "rule": issue.get("rule", ""), "project": issue.get("project", ""), "component": issue.get("component", ""), "file": file_path, "line": issue.get("line", ""), "hash": issue.get("hash", ""), "effort": issue.get("effort", ""), "debt": issue.get("debt", ""), "author": issue.get("author", ""), "assignee": issue.get("assignee", ""), "creation_date": issue.get("creationDate", ""), "update_date": issue.get("updateDate", ""), "clean_code_attribute": issue.get("cleanCodeAttribute", ""), "clean_code_attribute_category": issue.get("cleanCodeAttributeCategory", ""), "impacts": flatten_impacts(issue), "tags": "|".join(issue.get("tags", [])), "text_range": flatten_text_range(issue), "message": sanitize_text(issue.get("message", "")), } ) def export_hotspots(): with open(HOTSPOTS_CSV, "w", newline="", encoding="utf-8") as simple_file, open( HOTSPOTS_DETAILED_CSV, "w", newline="", encoding="utf-8" ) as detailed_file: simple_writer = csv.DictWriter(simple_file, fieldnames=HOTSPOT_FIELDS) detailed_writer = csv.DictWriter(detailed_file, fieldnames=HOTSPOT_DETAILED_FIELDS) simple_writer.writeheader() detailed_writer.writeheader() for hotspots in paginate_hotspots(): for hotspot in hotspots: file_path = component_path(hotspot) base_row = { "status": hotspot.get("status", ""), "vulnerability_probability": hotspot.get("vulnerabilityProbability", ""), "security_category": hotspot.get("securityCategory", ""), "file": file_path, "line": hotspot.get("line", ""), "rule_key": hotspot.get("ruleKey", ""), "message": sanitize_text(hotspot.get("message", "")), } simple_writer.writerow(base_row) detailed_writer.writerow( { "key": hotspot.get("key", ""), "project": hotspot.get("project", ""), "component": hotspot.get("component", ""), "file": file_path, "line": hotspot.get("line", ""), "status": hotspot.get("status", ""), "security_category": hotspot.get("securityCategory", ""), "vulnerability_probability": hotspot.get("vulnerabilityProbability", ""), "rule_key": hotspot.get("ruleKey", ""), "author": hotspot.get("author", ""), "creation_date": hotspot.get("creationDate", ""), "update_date": hotspot.get("updateDate", ""), "text_range": flatten_text_range(hotspot), "message": sanitize_text(hotspot.get("message", "")), } ) def export_coverage(): base_metrics = [ "coverage", "line_coverage", "branch_coverage", "lines_to_cover", "uncovered_lines", "conditions_to_cover", "uncovered_conditions", "new_coverage", "new_line_coverage", "new_branch_coverage", "new_lines_to_cover", "new_uncovered_lines", "new_conditions_to_cover", "new_uncovered_conditions", "ncloc", "complexity", "cognitive_complexity", ] detailed_metrics = list(base_metrics) for metric in OPTIONAL_FILE_METRICS: try: fetch_json_once( "/api/measures/component_tree", { "component": PROJECT, "metricKeys": metric, "qualifiers": "FIL", "ps": 1, "p": 1, }, ) detailed_metrics.append(metric) except Exception as exc: print(f"Metric ignorée (non supportée): {metric} — {exc}") with open(COVERAGE_CSV, "w", newline="", encoding="utf-8") as simple_file, open( COVERAGE_DETAILED_CSV, "w", newline="", encoding="utf-8" ) as detailed_file: simple_writer = csv.DictWriter(simple_file, fieldnames=COVERAGE_FIELDS) detailed_writer = csv.DictWriter(detailed_file, fieldnames=COVERAGE_DETAILED_FIELDS) simple_writer.writeheader() detailed_writer.writeheader() for components in paginate_files(detailed_metrics, "Coverage"): for component in components: measures = measure_map(component) file_path = component_path(component) simple_writer.writerow( { "file": file_path, "coverage": measures.get("coverage", ""), "uncovered_lines": measures.get("uncovered_lines", ""), "uncovered_conditions": measures.get("uncovered_conditions", ""), "lines_to_cover": measures.get("lines_to_cover", ""), } ) detailed_writer.writerow( { "file": file_path, "name": component.get("name", ""), "qualifier": component.get("qualifier", ""), "language": component.get("language", ""), "coverage": measures.get("coverage", ""), "line_coverage": measures.get("line_coverage", ""), "branch_coverage": measures.get("branch_coverage", ""), "lines_to_cover": measures.get("lines_to_cover", ""), "uncovered_lines": measures.get("uncovered_lines", ""), "conditions_to_cover": measures.get("conditions_to_cover", ""), "uncovered_conditions": measures.get("uncovered_conditions", ""), "new_coverage": measures.get("new_coverage", ""), "new_line_coverage": measures.get("new_line_coverage", ""), "new_branch_coverage": measures.get("new_branch_coverage", ""), "new_lines_to_cover": measures.get("new_lines_to_cover", ""), "new_uncovered_lines": measures.get("new_uncovered_lines", ""), "new_conditions_to_cover": measures.get("new_conditions_to_cover", ""), "new_uncovered_conditions": measures.get("new_uncovered_conditions", ""), "ncloc": measures.get("ncloc", ""), "complexity": measures.get("complexity", ""), "cognitive_complexity": measures.get("cognitive_complexity", ""), "coverage_line_hist_data": measures.get("coverage_line_hist_data", ""), "branch_coverage_hits_data": measures.get("branch_coverage_hits_data", ""), "conditions_by_line": measures.get("conditions_by_line", ""), "covered_conditions_by_line": measures.get("covered_conditions_by_line", ""), } ) def export_duplications(): duplication_metrics = [ "duplicated_lines_density", "duplicated_lines", "duplicated_blocks", "new_duplicated_lines", "new_duplicated_blocks", "new_duplicated_lines_density", "ncloc", ] with open(DUPLICATION_CSV, "w", newline="", encoding="utf-8") as simple_file, open( DUPLICATION_DETAILED_CSV, "w", newline="", encoding="utf-8" ) as detailed_file: simple_writer = csv.DictWriter(simple_file, fieldnames=DUPLICATION_FIELDS) detailed_writer = csv.DictWriter(detailed_file, fieldnames=DUPLICATION_DETAILED_FIELDS) simple_writer.writeheader() detailed_writer.writeheader() for components in paginate_files(duplication_metrics, "Duplications"): for component in components: measures = measure_map(component) file_path = component_path(component) base_row = { "file": file_path, "duplicated_lines_density": measures.get("duplicated_lines_density", ""), "duplicated_lines": measures.get("duplicated_lines", ""), "duplicated_blocks": measures.get("duplicated_blocks", ""), "new_duplicated_lines": measures.get("new_duplicated_lines", ""), "new_duplicated_blocks": measures.get("new_duplicated_blocks", ""), "new_duplicated_lines_density": measures.get("new_duplicated_lines_density", ""), "ncloc": measures.get("ncloc", ""), } simple_writer.writerow(base_row) detailed_writer.writerow( { **base_row, "name": component.get("name", ""), "qualifier": component.get("qualifier", ""), "language": component.get("language", ""), } ) def export_summary(): data = fetch_json( "/api/measures/component", { "component": PROJECT, "metricKeys": ",".join(PROJECT_METRICS), }, ) component = data.get("component", {}) quality_gate = fetch_json( "/api/qualitygates/project_status", { "projectKey": PROJECT, }, ).get("projectStatus", {}) summary = { "host": HOST, "project": PROJECT, "branch": BRANCH, "component": component.get("key", ""), "name": component.get("name", ""), "qualifier": component.get("qualifier", ""), "measures": measure_map(component), "quality_gate": quality_gate, } with open(SUMMARY_JSON, "w", encoding="utf-8") as summary_file: json.dump(summary, summary_file, indent=2, ensure_ascii=False) summary_file.write("\n") def main(): if not TOKEN: print("SONAR_TOKEN manquant", file=sys.stderr) sys.exit(1) export_issues() export_hotspots() export_coverage() export_duplications() export_summary() print( "Export terminé : " f"{ISSUES_CSV}, {ISSUES_DETAILED_CSV}, " f"{HOTSPOTS_CSV}, {HOTSPOTS_DETAILED_CSV}, " f"{COVERAGE_CSV}, {COVERAGE_DETAILED_CSV}, " f"{DUPLICATION_CSV}, {DUPLICATION_DETAILED_CSV}, {SUMMARY_JSON}" ) if __name__ == "__main__": main() PYEOF python3 /tmp/export_sonar.py