merge branche + changing conf structure for commmunity + fixing bug when running the same extraction again

Author: Nasser2003

algorithms/GraphAnalysis.py

@@ -76,15 +76,16 @@ def _run_graph_generation(self) -> None:
         final_fmt = out.final_file_format
 
         if gt.user_user:
-            if getattr(p, "is_community", False) and getattr(p, "community_file", ""):
-                df = pd.read_csv(p.community_file)
+            comm_cfg = getattr(p, "community", None)
+            if comm_cfg and getattr(comm_cfg, "is_community", False) and getattr(comm_cfg, "community_file", ""):
+                df = pd.read_csv(comm_cfg.community_file)
                 user_community_map = dict(
                     zip(df["user_id"].astype(int), df["community"].astype(int))
                 )
 
                 # ── automatical selection of strategy ──────────────────────────
-                strategy_name = getattr(p, "community_strategy")
-                batch_size = getattr(p, "community_batch_size")
+                strategy_name = getattr(comm_cfg, "community_strategy")
+                batch_size = getattr(comm_cfg, "community_batch_size")
 
                 if strategy_name == "batch":
                     strategy = CommunityUserBatchStrategy(user_community_map)
@@ -113,7 +114,7 @@ def _run_graph_generation(self) -> None:
                 fast_rt_threshold=p.fast_rt_threshold,
                 strategy=strategy,
                 is_community_run=is_comm,
-                community_file_path=getattr(p, "community_file", None) if is_comm else None,
+                community_file_path=getattr(comm_cfg, "community_file", None) if is_comm else None,
                 load_snapshot_status=p.load_snapshot.status if p.load_snapshot else False,
                 load_snapshot_tmp_path=p.load_snapshot.tmp_path if p.load_snapshot else ""
             )

algorithms/graph/extraction.py

@@ -386,6 +386,21 @@ def _process_work_item(self, worker_id: int, work_item,
 
     def run(self, checkpoint_every: int = 500,
             max_users: Optional[int] = None, n_workers: int = 4) -> None:
+        if self.load_snapshot_status and self.load_snapshot_tmp_path:
+            snapshot_basename = os.path.basename(
+                self.load_snapshot_tmp_path.rstrip(os.sep))
+            if snapshot_basename:
+                self.id = snapshot_basename
+
+        # Check if already complete
+        final_dir = os.path.join(self.output_file_path, self.id)
+        metadata_file = os.path.join(final_dir, "metadata.json")
+        if self.load_snapshot_status and os.path.exists(metadata_file):
+            self.logger.info(
+                f"Final output metadata.json found at '{metadata_file}'. "
+                "Extraction and merge have already completed for this snapshot. Skipping.")
+            return
+
         self.logger.info(
             f"[GraphGenerationUser] Starting extraction. "
             f"Run ID: {self.id}  Format: {self.file_format}")
@@ -405,10 +420,6 @@ def run(self, checkpoint_every: int = 500,
             skipped = self.strategy.exclude_users(processed)
             self.logger.info(
                 f"Excluded {skipped} users from strategy processing queue.")
-            snapshot_basename = os.path.basename(
-                self.load_snapshot_tmp_path.rstrip(os.sep))
-            if snapshot_basename:
-                self.id = snapshot_basename
 
         work_items = self.strategy.partition_work(
             collection, n_workers, max_users, self.logger)

config.py

@@ -89,14 +89,17 @@ class LoadSnapshot(BaseModel):
     status: bool = False
     tmp_path: str = ""
 
+class CommunityExtractionConfig(BaseModel):
+    is_community: bool = False
+    community_file: str = ""
+    community_strategy: str = "linear"
+    community_batch_size: int = 500
+
 class GraphParameters(BaseModel):
     checkpoint_every: int = 5000
     n_workers: int = 4
     fast_rt_threshold: int = 60
-    is_community: bool
-    community_file: str
-    community_strategy: str
-    community_batch_size: int
+    community: CommunityExtractionConfig = CommunityExtractionConfig()
     load_snapshot: Optional[LoadSnapshot] = LoadSnapshot()
     input: SourceInput
     graph_type: GraphTypes

properties/prop.ipazia.yaml

@@ -8,10 +8,11 @@ graph_generation:
     checkpoint_every: 100
     n_workers: 64
     fast_rt_threshold: 60
-    is_community: true
-    community_file: "/app/resources/communities_rp_0.4.csv"
-    community_strategy: "shard"   # "batch" | "shard" | "linear"
-    community_batch_size: 100
+    community:
+      is_community: true
+      community_file: "/app/resources/communities_rp_0.4.csv"
+      community_strategy: "shard"   # "batch" | "shard" | "linear"
+      community_batch_size: 100
     input:
       type: "json"
       conf:

properties/prop.local.yaml

@@ -8,10 +8,11 @@ graph_generation:
     checkpoint_every: 500
     n_workers: 4
     fast_rt_threshold: 60
-    is_community: false
-    community_file: "./resources/communities.csv"
-    community_strategy: "linear"   # "batch" | "shard" | "linear"
-    community_batch_size: 500
+    community:
+      is_community: false
+      community_file: "./resources/communities.csv"
+      community_strategy: "linear"   # "batch" | "shard" | "linear"
+      community_batch_size: 500
     input:
       type: "mongo"
       conf:

properties/prop.template.yaml

@@ -21,10 +21,11 @@ graph_generation:
     checkpoint_every: 5000
     n_workers: 4
     fast_rt_threshold: 60
-    is_community: false
-    community_file: "./resources/communities.csv"
-    community_strategy: "linear"   # "batch" | "shard" | "linear"
-    community_batch_size: 500
+    community:
+      is_community: false
+      community_file: "./resources/communities.csv"
+      community_strategy: "linear"   # "batch" | "shard" | "linear"
+      community_batch_size: 500
     input:
       type: "json" # "mongo", "json", "csv"
       conf:

properties/prop.yaml

@@ -11,10 +11,11 @@ graph_generation:
     checkpoint_every: 100
     n_workers: 8
     fast_rt_threshold: 60
-    is_community: true
-    community_file: "./resources/communities_rp_0.4.csv"
-    community_strategy: "shard"   # "batch" | "shard" | "linear"
-    community_batch_size: 100
+    community:
+      is_community: true
+      community_file: "./resources/communities_rp_0.4.csv"
+      community_strategy: "shard"   # "batch" | "shard" | "linear"
+      community_batch_size: 100
     input:
       type: "json"
       conf:
@@ -38,7 +39,7 @@ graph_generation:
       user_user: true
     load_snapshot:
       status: true
-      tmp_path: "./resources/tmp/48ca37c8507711f1abea9c29764a0d1b"
+      tmp_path: "./resources/tmp/9e5ab1a1507f11f1b8329c29764a0d1b"
     output:
       path: "./resources/"
       graph_file_name: