=
136 lines
4.7 KiB
Python
136 lines
4.7 KiB
Python
# Copyright (C) 2025 James Brotosky, Brandon Wickline
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import pandas as pd
|
|
import os
|
|
import ast
|
|
import re
|
|
|
|
|
|
def split_filepaths_grouped(df, col="filename", group_parts=3, min_parts=3):
|
|
def clean_split(path):
|
|
parts = os.path.normpath(path).split(os.sep)
|
|
# Remove leading empty strings caused by UNC paths
|
|
parts = [p for p in parts if p]
|
|
return parts
|
|
|
|
df = df.copy()
|
|
split_paths = df[col].apply(clean_split)
|
|
|
|
# Filter out paths with fewer than `min_parts` components
|
|
df = df[split_paths.apply(lambda parts: len(parts) >= min_parts)].copy()
|
|
split_paths = split_paths[df.index] # Update split_paths to match filtered df
|
|
|
|
df["group_key"] = split_paths.apply(lambda parts: os.sep.join(parts[:group_parts]))
|
|
grouped = df.groupby("group_key")
|
|
new_rows = []
|
|
|
|
for _, group_df in grouped:
|
|
paths = group_df[col].tolist()
|
|
split_parts = [clean_split(p) for p in paths]
|
|
|
|
def longest_common_prefix(paths):
|
|
if not paths:
|
|
return []
|
|
prefix = paths[0]
|
|
for path in paths[1:]:
|
|
prefix = [a for a, b in zip(prefix, path) if a == b]
|
|
if not prefix:
|
|
break
|
|
return prefix
|
|
|
|
common_prefix = longest_common_prefix(split_parts)
|
|
prefix_str = os.sep.join(common_prefix)
|
|
|
|
for i, parts in enumerate(split_parts):
|
|
filename = parts[-1]
|
|
middle = os.sep.join(parts[len(common_prefix):-1]) if len(parts) > len(common_prefix) + 1 else ""
|
|
row = group_df.iloc[i].copy()
|
|
row["longestcfp"] = prefix_str
|
|
row["middle"] = middle
|
|
row["filename_only"] = filename
|
|
new_rows.append(row)
|
|
|
|
return pd.DataFrame(new_rows).drop(columns=["group_key"])
|
|
|
|
def mask_from_csv(df, csv_path, filepath_col):
|
|
"""
|
|
Reads reviewed CSV of groups, keeps only files in approved groups.
|
|
"""
|
|
review_df = pd.read_csv(csv_path)
|
|
|
|
def parse_paths(val):
|
|
if isinstance(val, str):
|
|
try:
|
|
# Try to parse as a list
|
|
parsed = ast.literal_eval(val)
|
|
# If it's not a list, wrap it
|
|
return parsed if isinstance(parsed, list) else [parsed]
|
|
except (ValueError, SyntaxError):
|
|
# If parsing fails, treat it as a single path
|
|
return [val]
|
|
return [val]
|
|
|
|
review_df[filepath_col] = review_df[filepath_col].apply(parse_paths)
|
|
|
|
# Flatten all approved file paths into a set for masking
|
|
approved_files = set()
|
|
for paths in review_df[filepath_col]:
|
|
approved_files.update(paths)
|
|
|
|
# Keep only rows in df that are in approved_files
|
|
masked_df = df[df[filepath_col].isin(approved_files)].copy()
|
|
remainder = df[~df[filepath_col].isin(approved_files)].copy()
|
|
return remainder
|
|
|
|
def filter_and_drop(approved, eligiblepaths, min_hashes):
|
|
"""
|
|
Filters eligiblepaths to rows where all hashes are in approved,
|
|
then drops rows with fewer than min_hashes hashes.
|
|
"""
|
|
approved_hashes = set(approved['sha256'])
|
|
|
|
def all_hashes_approved(row):
|
|
return all(h in approved_hashes for h in row['sha256'])
|
|
|
|
filtered = eligiblepaths[eligiblepaths.apply(all_hashes_approved, axis=1)]
|
|
filtered = filtered[filtered['sha256'].apply(len) >= min_hashes]
|
|
|
|
return filtered
|
|
|
|
def inspect_parquet(path):
|
|
try:
|
|
df = pd.read_parquet(path)
|
|
print(f"✅ Successfully read: {path}")
|
|
print(f"📄 Columns: {df.columns.tolist()}")
|
|
print(f"🔢 Rows: {len(df)}")
|
|
return df
|
|
except Exception as e:
|
|
print(f"❌ Error reading {path}: {e}")
|
|
return pd.DataFrame()
|
|
|
|
|
|
def regulator(paths, case_insensitive=True):
|
|
"""
|
|
Build a regex pattern that matches any of the given Windows path fragments.
|
|
"""
|
|
escaped = [re.escape(p) for p in paths]
|
|
pattern = "(?:" + "|".join(escaped) + ")"
|
|
if case_insensitive:
|
|
pattern = "(?i)" + pattern # Add inline case-insensitive flag
|
|
print(f"Regulator is providing: {pattern}")
|
|
return pattern
|
|
|