mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
import csv
|
|
import os
|
|
from typing import Optional
|
|
|
|
from backend.pashub_fetcher.sharepoint_subfolders import SharepointSubfolders
|
|
from utilities.logger import setup_logger
|
|
from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
|
|
|
|
BASE_PATH = (
|
|
"Osmosis-ACD Projects/Sero-Clarion Housing/"
|
|
"Sero Project Documents/Property Folders"
|
|
)
|
|
ASSESSMENT_SUBFOLDER = "A. Assessment"
|
|
|
|
logger = setup_logger()
|
|
|
|
|
|
def build_canonical_filename(
|
|
uprn: str, address: str, postcode: str, original_name: str
|
|
) -> Optional[str]:
|
|
"""
|
|
Returns the canonical filename, or None if the file is already renamed.
|
|
|
|
Already-renamed: name starts with "{uprn}_".
|
|
Strips any existing address prefix (address+postcode first, then address alone)
|
|
before inserting the canonical prefix.
|
|
"""
|
|
if original_name.startswith(f"{uprn}_"):
|
|
return None
|
|
|
|
stem, ext = os.path.splitext(original_name)
|
|
stem_lower = stem.lower()
|
|
|
|
street = address.split(",")[0].strip()
|
|
prefixes = [
|
|
f"{address} {postcode}",
|
|
address,
|
|
f"{street} {postcode}",
|
|
street,
|
|
]
|
|
|
|
doc_name = stem
|
|
for prefix in prefixes:
|
|
if stem_lower.startswith(prefix.lower()):
|
|
doc_name = stem[len(prefix) :]
|
|
break
|
|
|
|
if doc_name.startswith(" - "):
|
|
doc_name = doc_name[3:]
|
|
elif doc_name.startswith(" _ "):
|
|
doc_name = doc_name[3:]
|
|
doc_name = doc_name.strip()
|
|
|
|
street_post = f"{street} {postcode}"
|
|
if doc_name:
|
|
return f"{uprn}_{street_post}_{doc_name}{ext}"
|
|
return f"{uprn}_{street_post}{ext}"
|
|
|
|
|
|
class SharepointRenamerOrchestrator:
|
|
def __init__(
|
|
self, sp_client: DomnaSharepointClient, csv_path: str, dry_run: bool = False
|
|
) -> None:
|
|
self._sp_client = sp_client
|
|
self._csv_path = csv_path
|
|
self._dry_run = dry_run
|
|
|
|
def run(self) -> None:
|
|
with open(self._csv_path, newline="", encoding="utf-8-sig") as f:
|
|
reader = csv.DictReader(f)
|
|
required = {"UPRN", "Address", "Postcode"}
|
|
if not reader.fieldnames or not required.issubset(set(reader.fieldnames)):
|
|
raise ValueError(
|
|
f"CSV missing required columns. Expected {required}, got {reader.fieldnames}"
|
|
)
|
|
|
|
for row in reader:
|
|
uprn = row["UPRN"].strip()
|
|
address = row["Address"].strip()
|
|
postcode = row["Postcode"].strip()
|
|
folder_path = (
|
|
f"{BASE_PATH}/{address}, {postcode}"
|
|
f"/{SharepointSubfolders.ASSESSMENT.value}/{ASSESSMENT_SUBFOLDER}"
|
|
)
|
|
self._process_folder(folder_path, uprn, address, postcode)
|
|
|
|
def _process_folder(
|
|
self, folder_path: str, uprn: str, address: str, postcode: str
|
|
) -> None:
|
|
try:
|
|
contents = self._sp_client.get_folders_in_path(folder_path)
|
|
except ValueError:
|
|
logger.warning(f"Missing folder for UPRN {uprn}: {folder_path}")
|
|
return
|
|
|
|
for item in contents.get("value", []):
|
|
if "folder" in item:
|
|
self._process_folder(
|
|
f"{folder_path}/{item['name']}", uprn, address, postcode
|
|
)
|
|
elif "file" in item:
|
|
original_name: str = item["name"]
|
|
if os.path.splitext(original_name)[1].lower() in {".jpg", ".heic"}:
|
|
continue
|
|
new_name = build_canonical_filename(
|
|
uprn, address, postcode, original_name
|
|
)
|
|
|
|
if new_name is None:
|
|
continue
|
|
|
|
if self._dry_run:
|
|
logger.info(
|
|
f'Would rename: "{original_name}" → "{new_name}" (UPRN: {uprn})'
|
|
)
|
|
else:
|
|
try:
|
|
self._sp_client.rename_file(item["id"], new_name)
|
|
logger.info(
|
|
f'Renamed: "{original_name}" → "{new_name}" (UPRN: {uprn})'
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f'Failed to rename "{original_name}" → "{new_name}" (UPRN: {uprn}): {e}'
|
|
)
|