108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
|
import os
|
||
|
|
import sys
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
import time
|
||
|
|
|
||
|
|
# Force stdout to be UTF-8 for Windows terminals
|
||
|
|
if hasattr(sys.stdout, 'reconfigure'):
|
||
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
||
|
|
|
||
|
|
NOW = time.time()
|
||
|
|
|
||
|
|
def get_git_timestamp(filepath):
|
||
|
|
# Normalize paths to use forward slashes for git compatibility
|
||
|
|
git_path = filepath.replace('\\', '/')
|
||
|
|
|
||
|
|
# 1. Check if the file exists on disk.
|
||
|
|
if not os.path.exists(filepath):
|
||
|
|
return 0
|
||
|
|
|
||
|
|
# 2. Check for local modifications (staged or unstaged)
|
||
|
|
try:
|
||
|
|
status_proc = subprocess.run(
|
||
|
|
['git', 'status', '--porcelain', '--', git_path],
|
||
|
|
capture_output=True, text=True, check=True
|
||
|
|
)
|
||
|
|
if status_proc.stdout.strip():
|
||
|
|
# Locally modified files are considered "edited just now"
|
||
|
|
return NOW
|
||
|
|
except subprocess.SubprocessError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# 3. Get the last commit timestamp
|
||
|
|
try:
|
||
|
|
log_proc = subprocess.run(
|
||
|
|
['git', 'log', '-1', '--format=%ct', '--', git_path],
|
||
|
|
capture_output=True, text=True, check=True
|
||
|
|
)
|
||
|
|
output = log_proc.stdout.strip()
|
||
|
|
if output.isdigit():
|
||
|
|
return int(output)
|
||
|
|
except subprocess.SubprocessError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# 4. Fallback to OS modification time if file exists but has no git commits yet
|
||
|
|
try:
|
||
|
|
return os.path.getmtime(filepath)
|
||
|
|
except OSError:
|
||
|
|
return 0
|
||
|
|
|
||
|
|
def check_dirty():
|
||
|
|
graph_path = os.path.join(os.path.dirname(__file__), 'graph.json')
|
||
|
|
if not os.path.exists(graph_path):
|
||
|
|
print(f"Error: graph.json not found at {graph_path}")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
with open(graph_path, 'r', encoding='utf-8') as f:
|
||
|
|
graph = json.load(f)
|
||
|
|
|
||
|
|
files_dict = graph.get('files', {})
|
||
|
|
timestamps = {}
|
||
|
|
|
||
|
|
# Calculate timestamps for all files
|
||
|
|
for filepath in files_dict.keys():
|
||
|
|
timestamps[filepath] = get_git_timestamp(filepath)
|
||
|
|
|
||
|
|
dirty_files = {}
|
||
|
|
|
||
|
|
# Check dependencies
|
||
|
|
for filepath, info in files_dict.items():
|
||
|
|
depends_on = info.get('depends_on', [])
|
||
|
|
file_ts = timestamps[filepath]
|
||
|
|
|
||
|
|
# If the file itself doesn't exist, it's not "dirty" in the sense of being out of date,
|
||
|
|
# but it is missing. However, to keep it clean, if it doesn't exist, we can skip or flag it.
|
||
|
|
# Let's say if a file exists, we check if its dependencies are newer.
|
||
|
|
if file_ts == 0:
|
||
|
|
# File is missing on disk
|
||
|
|
continue
|
||
|
|
|
||
|
|
file_dirty_deps = []
|
||
|
|
for dep in depends_on:
|
||
|
|
dep_ts = get_git_timestamp(dep)
|
||
|
|
if dep_ts > file_ts:
|
||
|
|
file_dirty_deps.append((dep, dep_ts, file_ts))
|
||
|
|
|
||
|
|
if file_dirty_deps:
|
||
|
|
dirty_files[filepath] = file_dirty_deps
|
||
|
|
|
||
|
|
if dirty_files:
|
||
|
|
print("\n[DIRTY] 发现以下下游文件落后于其上游依赖：")
|
||
|
|
for filepath, deps in dirty_files.items():
|
||
|
|
print(f"\n* {filepath}")
|
||
|
|
for dep, dep_ts, file_ts in deps:
|
||
|
|
dep_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(dep_ts))
|
||
|
|
file_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(file_ts))
|
||
|
|
print(f" [依赖于] {dep}")
|
||
|
|
print(f" (上游最后提交: {dep_time} > 下游最后提交: {file_time})")
|
||
|
|
print("\n请跟进并更新上述下游文件，重新 commit 即可刷新基线并自动消脏。\n")
|
||
|
|
return False
|
||
|
|
else:
|
||
|
|
print("\n[CLEAN] 所有文件均已与依赖同步，无脏文件。\n")
|
||
|
|
return True
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
success = check_dirty()
|
||
|
|
sys.exit(0 if success else 1)
|