chore: unify include analysis loading logic and support direct URLs

Author: dsanders11

add_to_remove_include.py

@@ -8,7 +8,7 @@
 
 from common import IgnoresConfiguration, IncludeChange
 from filter_include_changes import Change, filter_changes
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from list_includers import list_includers
 from list_transitive_includes import list_transitive_includes
 from typing import Dict, Iterator, List, Set, Tuple
@@ -97,7 +97,7 @@ def main():
     )
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         help="The include analysis output to use.",
     )
     parser.add_argument("filename", help="File that has the include to be removed.")
@@ -111,7 +111,7 @@ def main():
     args = parser.parse_args()
 
     try:
-        include_analysis = parse_raw_include_analysis_output(args.include_analysis_output.read())
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

extract_archived_include_analysis.py

@@ -3,20 +3,10 @@
 import argparse
 import logging
 import os
-import re
 import sys
 import urllib.request
 
-DATA_REGEX = re.compile(r".*<script>\n?(data = .*?)<\/script>", re.DOTALL)
-
-
-def extract_include_analysis(contents: str) -> str:
-    data_match = DATA_REGEX.match(contents)
-
-    if data_match:
-        return data_match.group(1).strip()
-
-    return ""
+from utils import extract_include_analysis
 
 
 def main():

extract_include_analysis_edges.py

@@ -6,16 +6,14 @@
 import os
 import re
 import sys
-import urllib.request
 from datetime import datetime
 
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from suggest_include_changes import filter_filenames
 from utils import (
     get_include_analysis_edge_expanded_sizes,
     get_include_analysis_edge_prevalence,
     get_include_analysis_edges_centrality,
-    get_latest_include_analysis,
 )
 
 
@@ -65,7 +63,7 @@ def main():
     parser = argparse.ArgumentParser(description="Extract include edges from include analysis, with filtering")
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
         help="The include analysis output to use.",
     )
@@ -93,14 +91,8 @@ def main():
         level=logging.DEBUG if args.verbose else logging.WARNING if args.quiet else logging.INFO,
     )
 
-    # If the user specified an include analysis output file, use that instead of fetching it
-    if args.include_analysis_output:
-        raw_include_analysis = args.include_analysis_output.read()
-    else:
-        raw_include_analysis = get_latest_include_analysis()
-
     try:
-        include_analysis = parse_raw_include_analysis_output(raw_include_analysis)
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

include_analysis.py

@@ -1,7 +1,10 @@
 import json
 import re
+import sys
+import urllib.request
 from typing import Dict, List, Optional, TypedDict
 
+DATA_REGEX = re.compile(r".*<script>\n?(data = .*?)<\/script>", re.DOTALL)
 GENERATED_FILE_PREFIX_REGEX = re.compile(r"^(out/[\w-]+/gen/).*$")
 SYSROOT_REGEX = re.compile(r"^(build/linux/[\w-]+/)usr/include/([\w-]+)/sys/.*$")
 
@@ -124,3 +127,41 @@ def parse_raw_include_analysis_output(output: str) -> Optional[IncludeAnalysisOu
     parsed_output["sysroot_platform"] = sysroot_platform
 
     return parsed_output
+
+
+def get_latest_include_analysis():
+    response = urllib.request.urlopen(
+        "https://commondatastorage.googleapis.com/chromium-browser-clang/include-analysis.js"
+    )
+
+    return response.read().decode("utf8")
+
+
+def extract_include_analysis(contents: str) -> str:
+    data_match = DATA_REGEX.match(contents)
+
+    if data_match:
+        return data_match.group(1).strip()
+
+    return ""
+
+
+def load_include_analysis(include_analysis_path: Optional[str]) -> IncludeAnalysisOutput:
+    # If the user specified an include analysis output file, use that instead of fetching it
+    if include_analysis_path:
+        if include_analysis_path.startswith("https://"):
+            include_analysis_response = urllib.request.urlopen(include_analysis_path)
+            include_analysis_contents = include_analysis_response.read().decode("utf8")
+            raw_include_analysis = extract_include_analysis(include_analysis_contents)
+
+            if not raw_include_analysis:
+                raise RuntimeError(f"Could not extract include analysis from {include_analysis_path}")
+        elif include_analysis_path == "-":
+            raw_include_analysis = sys.stdin.read()
+        else:
+            with open(include_analysis_path, "r") as f:
+                raw_include_analysis = f.read()
+    else:
+        raw_include_analysis = get_latest_include_analysis()
+
+    return parse_raw_include_analysis_output(raw_include_analysis)

include_analysis_diff.py

@@ -10,9 +10,13 @@
 from datetime import datetime
 
 from extract_archived_include_analysis import extract_include_analysis
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import (
+    IncludeAnalysisOutput,
+    ParseError,
+    load_include_analysis,
+    parse_raw_include_analysis_output,
+)
 from suggest_include_changes import filter_filenames
-from utils import get_latest_include_analysis
 
 CHROMIUM_INCLUDE_ANALYSIS_BASE_URL = "https://commondatastorage.googleapis.com/chromium-browser-clang"
 HREF_REGEX = re.compile(r"<a href=\"(.*?)\">", re.DOTALL)
@@ -220,13 +224,13 @@ def main():
     )
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
         help="The include analysis output to use.",
     )
     parser.add_argument(
         "previous_include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
         help="The previous include analysis output to use. If provided, the diff will only be between these two files.",
     )
@@ -264,14 +268,8 @@ def main():
         level=logging.DEBUG if args.verbose else logging.WARNING if args.quiet else logging.INFO,
     )
 
-    # If the user specified an include analysis output file, use that instead of fetching it
-    if args.include_analysis_output:
-        raw_include_analysis = args.include_analysis_output.read()
-    else:
-        raw_include_analysis = get_latest_include_analysis()
-
     try:
-        include_analysis = parse_raw_include_analysis_output(raw_include_analysis)
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")
@@ -280,10 +278,8 @@ def main():
         return 2
 
     if args.previous_include_analysis_output:
-        raw_previous_include_analysis = args.previous_include_analysis_output.read()
-
         try:
-            previous_include_analysis = parse_raw_include_analysis_output(raw_previous_include_analysis)
+            previous_include_analysis = load_include_analysis(args.previous_include_analysis_output)
         except ParseError as e:
             message = str(e)
             print("error: Could not parse include analysis output file")

list_includers.py

@@ -8,7 +8,7 @@
 
 from common import IgnoresConfiguration, IncludeChange
 from filter_include_changes import Change, filter_changes
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from typing import Iterator, List, Tuple
 from utils import (
     get_include_analysis_edges_centrality,
@@ -17,7 +17,6 @@
     get_include_analysis_edge_includer_size,
     get_include_analysis_edge_prevalence,
     get_include_analysis_edge_sizes,
-    get_latest_include_analysis,
     load_config,
     normalize_include_path,
 )
@@ -106,7 +105,7 @@ def main():
     parser = argparse.ArgumentParser(description="List includers of a file")
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
         help="The include analysis output to use.",
     )
@@ -136,14 +135,8 @@ def main():
     parser.add_argument("--verbose", action="store_true", default=False, help="Enable verbose logging.")
     args = parser.parse_args()
 
-    # If the user specified an include analysis output file, use that instead of fetching it
-    if args.include_analysis_output:
-        raw_include_analysis = args.include_analysis_output.read()
-    else:
-        raw_include_analysis = get_latest_include_analysis()
-
     try:
-        include_analysis = parse_raw_include_analysis_output(raw_include_analysis)
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

list_transitive_includes.py

@@ -9,7 +9,7 @@
 
 from common import FilteredIncludeChangeList, IgnoresConfiguration, IncludeChange
 from filter_include_changes import Change, filter_changes
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from typing import Dict, Iterator, List, Tuple
 from utils import (
     get_include_analysis_edges_centrality,
@@ -19,7 +19,6 @@
     get_include_analysis_edge_prevalence,
     get_include_analysis_edge_sizes,
     get_include_file_size,
-    get_latest_include_analysis,
     load_config,
     normalize_include_path,
 )
@@ -141,7 +140,7 @@ def main():
     parser = argparse.ArgumentParser(description="List transitive (and direct) includes of a file")
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
         help="The include analysis output to use.",
     )
@@ -179,14 +178,8 @@ def main():
         print("error: --apply-changes option requires --include-changes")
         return 1
 
-    # If the user specified an include analysis output file, use that instead of fetching it
-    if args.include_analysis_output:
-        raw_include_analysis = args.include_analysis_output.read()
-    else:
-        raw_include_analysis = get_latest_include_analysis()
-
     try:
-        include_analysis = parse_raw_include_analysis_output(raw_include_analysis)
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

minimum_edge_cut.py

@@ -8,13 +8,9 @@
 
 import networkx as nx
 
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from typing import Iterator, Tuple
-from utils import (
-    create_graph_from_include_analysis,
-    get_include_analysis_edge_prevalence,
-    get_latest_include_analysis,
-)
+from utils import create_graph_from_include_analysis, get_include_analysis_edge_prevalence
 
 
 def minimum_edge_cut(
@@ -69,9 +65,9 @@ def main():
     parser = argparse.ArgumentParser(description="Find the minimum edge cut between two files")
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         nargs="?",
-        help="The include analysis output to use.",
+        help="The include analysis output to use (can be a file path or URL). If not specified, pulls the latest.",
     )
     parser.add_argument("source", help="Source file.")
     parser.add_argument("target", help="Target file.")
@@ -86,14 +82,8 @@ def main():
     parser.add_argument("--verbose", action="store_true", default=False, help="Enable verbose logging.")
     args = parser.parse_args()
 
-    # If the user specified an include analysis output file, use that instead of fetching it
-    if args.include_analysis_output:
-        raw_include_analysis = args.include_analysis_output.read()
-    else:
-        raw_include_analysis = get_latest_include_analysis()
-
     try:
-        include_analysis = parse_raw_include_analysis_output(raw_include_analysis)
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

recalculate_expanded_sizes.py

@@ -11,10 +11,10 @@
 import time
 from itertools import batched
 
-from common import FilteredIncludeChangeList, IgnoresConfiguration, IncludeChange
+from common import FilteredIncludeChangeList, IgnoresConfiguration
 from filter_include_changes import Change, filter_changes
 from list_transitive_includes import list_transitive_includes
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from typing import Callable, Dict, Iterator, List, Tuple
 from utils import get_worker_count, load_config
 
@@ -180,7 +180,7 @@ def main():
     )
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         help="The include analysis output to use.",
     )
     parser.add_argument(
@@ -198,7 +198,7 @@ def main():
     args = parser.parse_args()
 
     try:
-        include_analysis = parse_raw_include_analysis_output(args.include_analysis_output.read())
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")

set_edge_weights.py

@@ -9,7 +9,7 @@
 
 from common import IgnoresConfiguration, IncludeChange
 from filter_include_changes import Change, filter_changes
-from include_analysis import IncludeAnalysisOutput, ParseError, parse_raw_include_analysis_output
+from include_analysis import IncludeAnalysisOutput, ParseError, load_include_analysis
 from utils import (
     get_include_analysis_edges_centrality,
     get_include_analysis_edge_expanded_sizes,
@@ -83,7 +83,7 @@ def main():
     )
     parser.add_argument(
         "include_analysis_output",
-        type=argparse.FileType("r"),
+        type=str,
         help="The include analysis output to use.",
     )
     parser.add_argument(
@@ -101,7 +101,7 @@ def main():
     args = parser.parse_args()
 
     try:
-        include_analysis = parse_raw_include_analysis_output(args.include_analysis_output.read())
+        include_analysis = load_include_analysis(args.include_analysis_output)
     except ParseError as e:
         message = str(e)
         print("error: Could not parse include analysis output file")