#!/usr/bin/env bash set -euo pipefail cat > /tmp/export_sonar.py << 'PYEOF' import base64 import csv import fnmatch import json import os import sys import time import urllib.parse import urllib.request TOKEN = os.getenv("SONAR_TOKEN", "squ_ba36e0dab96113b565300e34c1fef4e984bdc203") HOST = os.getenv("SONAR_HOST", "https://sonar.taaazzz-prog.fr").rstrip("/") PROJECT = os.getenv("SONAR_PROJECT", "iliacloud") BRANCH = os.getenv("SONAR_BRANCH", "").strip() PAGE_SIZE = int(os.getenv("SONAR_PAGE_SIZE", "100")) VERBOSE_SCOPE = os.getenv("SONAR_VERBOSE_SCOPE", "").strip() == "1" ISSUES_CSV = "sonar-issues.csv" ISSUES_DETAILED_CSV = "sonar-issues-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" HOTSPOT_CSV = "sonar-hotspots.csv" HOTSPOT_DETAILED_CSV = "sonar-hotspots-detailed.csv" 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", ] 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", ] HOTSPOT_FIELDS = [ "vulnerability_probability", "security_category", "status", "resolution", "file", "line", "message", ] HOTSPOT_DETAILED_FIELDS = [ "key", "vulnerability_probability", "security_category", "status", "resolution", "rule", "component", "file", "line", "hash", "author", "assignee", "creation_date", "update_date", "text_range", "message", ] 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", "security_hotspots", "security_hotspots_reviewed", ] OPTIONAL_FILE_METRICS = [ "coverage_line_hist_data", "branch_coverage_hits_data", "conditions_by_line", "covered_conditions_by_line", ] def load_sonar_scope(): sources = ["src", "public"] exclusions = [] props_path = os.path.join(os.getcwd(), "sonar-project.properties") if not os.path.exists(props_path): return sources, exclusions props = {} current_key = None current_value = [] with open(props_path, encoding="utf-8") as props_file: for raw_line in props_file: line = raw_line.strip() if not line or line.startswith("#"): continue if current_key is not None: if line.endswith("\\"): current_value.append(line[:-1].strip()) continue current_value.append(line) props[current_key] = "".join(current_value).strip() current_key = None current_value = [] continue if "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if value.endswith("\\"): current_key = key current_value = [value[:-1].strip()] continue props[key] = value raw_sources = props.get("sonar.sources", "") if raw_sources: sources = [part.strip().strip("/") for part in raw_sources.split(",") if part.strip()] raw_exclusions = props.get("sonar.exclusions", "") if raw_exclusions: exclusions = [part.strip() for part in raw_exclusions.split(",") if part.strip()] return sources, exclusions SOURCE_ROOTS, SOURCE_EXCLUSIONS = load_sonar_scope() 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 urllib.error.HTTPError as exc: if exc.code == 403: raise RuntimeError( f"Accès refusé (403) — le token doit avoir Browse + See Source Code sur le projet : {url}" ) from exc if attempt == retries: raise RuntimeError(f"Échec Sonar API après {retries} tentatives: {url}") from exc time.sleep(backoff * attempt) 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 component_path(component): path = component.get("path") if path: return path key = component.get("key", "") return key.split(":", 1)[-1] def normalize_repo_path(path): return sanitize_text(path).replace("\\", "/").lstrip("./") def is_path_in_scope(path): normalized = normalize_repo_path(path) if not normalized: return False if SOURCE_ROOTS and not any( normalized == root or normalized.startswith(root + "/") for root in SOURCE_ROOTS ): return False return not any(fnmatch.fnmatch(normalized, pattern) for pattern in SOURCE_EXCLUSIONS) def issue_in_scope(issue): project_key = sanitize_text(issue.get("project", "")) component_key = sanitize_text(issue.get("component", "")) file_path = component_key.split(":", 1)[-1] if ":" in component_key else component_key if project_key and project_key != PROJECT: return False return is_path_in_scope(file_path) def hotspot_in_scope(hotspot): component_key = sanitize_text(hotspot.get("component", "")) file_path = component_key.split(":", 1)[-1] if ":" in component_key else component_key return is_path_in_scope(file_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 fetched = 0 while fetched < total: data = fetch_json( "/api/issues/search", { "projectKeys": PROJECT, "statuses": "OPEN,CONFIRMED,REOPENED", "additionalFields": "_all", "ps": PAGE_SIZE, "p": page, }, ) issues = data.get("issues", []) total = data.get("total", 0) fetched += len(issues) yield issues page += 1 def paginate_files(metric_keys, label="Files"): """Pagine component_tree en essayant d'abord qualifiers=FIL, puis sans qualifiers (fallback) si 403, en filtrant les FIL côté Python.""" use_qualifier = True page = 1 total = 1 exported = 0 try: # Test préliminaire pour détecter 403 rapidement test_params = { "component": PROJECT, "metricKeys": metric_keys[0], "qualifiers": "FIL", "ps": 1, "p": 1, } fetch_json("/api/measures/component_tree", test_params) except RuntimeError as exc: if "403" in str(exc): print(f"⚠️ qualifiers=FIL refusé (403), tentative sans filtre qualifier...") use_qualifier = False else: raise while exported < total: params = { "component": PROJECT, "metricKeys": ",".join(metric_keys), "ps": PAGE_SIZE, "p": page, } if use_qualifier: params["qualifiers"] = "FIL" data = fetch_json("/api/measures/component_tree", params) all_components = data.get("components", []) components = [ c for c in all_components if use_qualifier or c.get("qualifier") == "FIL" ] total = data.get("paging", {}).get("total", 0) exported += len(all_components) print(f"{label} page {page} — {exported}/{total}") yield components page += 1 def export_issues(): filtered_out = 0 exported = 0 page = 1 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(): page_exported = 0 page_filtered = 0 for issue in issues: if not issue_in_scope(issue): filtered_out += 1 page_filtered += 1 continue component_key = issue.get("component", "") file_path = component_key.split(":", 1)[-1] if ":" in component_key else component_key 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", "")), } ) exported += 1 page_exported += 1 print( f"Issues page {page} — exportées: {page_exported}" + (f", hors périmètre: {page_filtered}" if VERBOSE_SCOPE and page_filtered else "") ) page += 1 if VERBOSE_SCOPE and filtered_out: print(f"Issues hors périmètre ignorées: {filtered_out}") print(f"Issues exportées: {exported}") 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, label="Coverage"): for component in components: file_path = component_path(component) if not is_path_in_scope(file_path): continue measures = measure_map(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, label="Duplications"): for component in components: file_path = component_path(component) if not is_path_in_scope(file_path): continue measures = measure_map(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 paginate_hotspots(): page = 1 total = 1 fetched = 0 while fetched < total: data = fetch_json( "/api/hotspots/search", { "projectKey": PROJECT, "statuses": "TO_REVIEW,REVIEWED", "ps": PAGE_SIZE, "p": page, }, ) hotspots = data.get("hotspots", []) total = data.get("paging", {}).get("total", 0) fetched += len(hotspots) yield hotspots page += 1 def export_hotspots(): filtered_out = 0 exported = 0 page = 1 with open(HOTSPOT_CSV, "w", newline="", encoding="utf-8") as simple_file, open( HOTSPOT_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(): page_exported = 0 page_filtered = 0 for hotspot in hotspots: if not hotspot_in_scope(hotspot): filtered_out += 1 page_filtered += 1 continue component_key = hotspot.get("component", "") file_path = component_key.split(":", 1)[-1] if ":" in component_key else component_key simple_writer.writerow( { "vulnerability_probability": hotspot.get("vulnerabilityProbability", ""), "security_category": hotspot.get("securityCategory", ""), "status": hotspot.get("status", ""), "resolution": hotspot.get("resolution", ""), "file": file_path, "line": hotspot.get("line", ""), "message": sanitize_text(hotspot.get("message", "")), } ) detailed_writer.writerow( { "key": hotspot.get("key", ""), "vulnerability_probability": hotspot.get("vulnerabilityProbability", ""), "security_category": hotspot.get("securityCategory", ""), "status": hotspot.get("status", ""), "resolution": hotspot.get("resolution", ""), "rule": hotspot.get("ruleKey", ""), "component": hotspot.get("component", ""), "file": file_path, "line": hotspot.get("line", ""), "hash": hotspot.get("hash", ""), "author": hotspot.get("author", ""), "assignee": hotspot.get("assignee", ""), "creation_date": hotspot.get("creationDate", ""), "update_date": hotspot.get("updateDate", ""), "text_range": flatten_text_range(hotspot), "message": sanitize_text(hotspot.get("message", "")), } ) exported += 1 page_exported += 1 print( f"Hotspots page {page} — exportés: {page_exported}" + (f", hors périmètre: {page_filtered}" if VERBOSE_SCOPE and page_filtered else "") ) page += 1 if VERBOSE_SCOPE and filtered_out: print(f"Hotspots hors périmètre ignorés: {filtered_out}") print(f"Hotspots exportés: {exported}") def export_summary(): data = fetch_json( "/api/measures/component", { "component": PROJECT, "metricKeys": ",".join(PROJECT_METRICS), }, ) component = data.get("component", {}) summary = { "host": HOST, "project": PROJECT, "branch": BRANCH, "component": component.get("key", ""), "name": component.get("name", ""), "qualifier": component.get("qualifier", ""), "measures": measure_map(component), } 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) exports = [ ("issues", export_issues), ("hotspots", export_hotspots), ("coverage", export_coverage), ("duplications", export_duplications), ("summary", export_summary), ] done = [] for name, fn in exports: try: fn() done.append(name) except Exception as exc: print(f"⚠️ Export '{name}' ignoré : {exc}", file=sys.stderr) print( "Export terminé : " f"{ISSUES_CSV}, {ISSUES_DETAILED_CSV}, " f"{HOTSPOT_CSV}, {HOTSPOT_DETAILED_CSV}, " f"{COVERAGE_CSV}, {COVERAGE_DETAILED_CSV}, " f"{DUPLICATION_CSV}, {DUPLICATION_DETAILED_CSV}, {SUMMARY_JSON}" ) print(f"Exports réussis : {', '.join(done)}") if __name__ == "__main__": main() PYEOF python3 /tmp/export_sonar.py