Tak — to bardzo ułatwi szukanie duplikatów, bo zamiast zgadywać intencję z kodu, dajesz algorytmowi krótki, ustrukturyzowany opis. I tak: można to zaadaptować do różnych poziomów złożoności:
linia kodu → scope:line
blok kodu → scope:block
funkcja → scope:function
metoda → scope:method
klasa → scope:class
plik → scope:file
moduł → scope:module
cały projekt → scope:project
Najważniejsze: każdy poziom dostaje tę samą sygnaturę intencji, tylko z innym scope.
Przykłady:
# @intent normalize:token !p3 @parser scope:line# @intent parse:extensions !p2 @cli in:raw out:list scope:function# @intent manage:hash_cache !p2 @cache fx:read,write scope:class# @intent analyze:duplication !p1 @project in:source_tree out:DuplicationMap scope:projectPoniżej masz gotowy moduł funkcji do pierwszego MVP.
from __future__ import annotations
import hashlib
import re
import shlex
from collections import defaultdict
from dataclasses import dataclass, field
from itertools import combinations
from typing import Iterable
VALID_SCOPES = {
"line",
"block",
"function",
"method",
"class",
"file",
"module",
"package",
"project",
}
ACTION_SYNONYMS = {
"validate": {"validate", "check", "verify", "ensure", "authorize", "guard"},
"build": {"build", "create", "make", "construct", "generate", "assemble"},
"parse": {"parse", "read", "load", "deserialize", "extract"},
"compare": {"compare", "diff", "match"},
"scan": {"scan", "collect", "discover", "find"},
"render": {"render", "format", "serialize", "export"},
"persist": {"persist", "save", "write", "store"},
"transform": {"transform", "normalize", "convert", "map"},
"detect": {"detect", "identify", "recognize", "classify"},
"calculate": {"calculate", "compute", "count", "measure", "estimate"},
}
STOP_WORDS = {
"a",
"an",
"the",
"to",
"from",
"of",
"for",
"before",
"after",
"with",
"without",
"and",
"or",
"by",
}
@dataclass(frozen=True)
class IntentDescriptor:
"""
Surowy opis intencji po sparsowaniu tagu RIDL.
Przykład tagu:
# @intent validate:user_permission !p1 @security in:user,resource out:allowed fx:read no:write #auth
"""
action: str
object: str
priority: int = 3
scope: str = "block"
domain: str = ""
inputs: tuple[str, ...] = field(default_factory=tuple)
outputs: tuple[str, ...] = field(default_factory=tuple)
effects: tuple[str, ...] = field(default_factory=tuple)
constraints: tuple[str, ...] = field(default_factory=tuple)
algorithms: tuple[str, ...] = field(default_factory=tuple)
tags: tuple[str, ...] = field(default_factory=tuple)
relations: tuple[str, ...] = field(default_factory=tuple)
intent_id: str = ""
raw: str = ""
@dataclass(frozen=True)
class IntentRecord:
"""
Intencja przypięta do konkretnego miejsca w kodzie.
"""
descriptor: IntentDescriptor
file_path: str
start_line: int
end_line: int
owner: str = ""
@dataclass(frozen=True)
class IntentSignature:
"""
Znormalizowana sygnatura do porównywania algorytmicznego.
"""
block_id: str
file_path: str
start_line: int
end_line: int
scope: str
action: str
object: str
domain: str
priority: int
features: frozenset[str]
exact_hash: str
raw: str = ""
@dataclass(frozen=True)
class IntentDuplicatePair:
left_id: str
right_id: str
similarity: float
reason: dict[str, object]
def clean_comment_line(line: str) -> str:
"""
Usuwa popularne prefiksy komentarzy:
#, //, --, ;, /*, */, *, <!--, -->
"""
text = line.strip()
if text.startswith("<!--"):
text = text[4:].strip()
if text.endswith("-->"):
text = text[:-3].strip()
if text.startswith("/*"):
text = text[2:].strip()
if text.endswith("*/"):
text = text[:-2].strip()
while text.startswith("*"):
text = text[1:].strip()
for prefix in ("#", "//", "--", ";"):
if text.startswith(prefix):
return text[len(prefix):].strip()
return text
def split_csv(value: str) -> tuple[str, ...]:
"""
Rozdziela wartości typu:
in:user,resource
fx:read,write
"""
value = value.strip()
if not value:
return ()
if value.lower() in {"none", "null", "empty", "-"}:
return ()
parts = [part.strip() for part in value.split(",")]
return tuple(part for part in parts if part)
def parse_key_value(token: str) -> tuple[str, str] | None:
"""
Obsługuje:
in:user,resource
in=user,resource
action:validate
action=validate
"""
match = re.match(r"^([A-Za-z_][A-Za-z0-9_-]*)(:|=)(.+)$", token)
if not match:
return None
key = match.group(1).strip().lower()
value = match.group(3).strip()
return key, value
def parse_priority(token: str) -> int | None:
"""
Obsługuje:
!p1
!p2
p1
priority:1
"""
token = token.strip().lower()
if token.startswith("!p"):
raw = token[2:]
elif token.startswith("p") and token[1:].isdigit():
raw = token[1:]
else:
return None
if not raw.isdigit():
return None
value = int(raw)
if value < 1:
return 1
if value > 5:
return 5
return value
def parse_intent_tag_line(
line: str,
*,
default_scope: str = "block",
) -> IntentDescriptor | None:
"""
Parsuje pojedynczy tag RIDL.
Format główny:
@intent action:object !p1 @domain in:a,b out:x fx:read no:write ~lsh #tag scope:function
Przykład:
# @intent validate:user_permission !p1 @security in:user,resource out:allowed fx:read no:write #auth
"""
cleaned = clean_comment_line(line)
if "@intent" not in cleaned:
return None
payload = cleaned.split("@intent", 1)[1].strip()
if not payload:
return None
tokens = shlex.split(payload, comments=False, posix=True)
action = ""
object_name = ""
priority = 3
scope = default_scope
domain = ""
inputs: list[str] = []
outputs: list[str] = []
effects: list[str] = []
constraints: list[str] = []
algorithms: list[str] = []
tags: list[str] = []
relations: list[str] = []
intent_id = ""
positional: list[str] = []
for token in tokens:
parsed_priority = parse_priority(token)
if parsed_priority is not None:
priority = parsed_priority
continue
if token.startswith("@") and len(token) > 1:
domain = token[1:]
continue
if token.startswith("#") and len(token) > 1:
tags.append(token[1:])
continue
if token.startswith("~") and len(token) > 1:
algorithms.append(token[1:])
continue
key_value = parse_key_value(token)
if key_value:
key, value = key_value
if key in {"action", "act"}:
action = value
elif key in {"object", "obj", "target"}:
object_name = value
elif key in {"domain"}:
domain = value
elif key in {"priority", "p"}:
if value.isdigit():
priority = max(1, min(5, int(value)))
elif key in {"scope"}:
scope = value if value in VALID_SCOPES else default_scope
elif key in {"in", "input", "inputs"}:
inputs.extend(split_csv(value))
elif key in {"out", "output", "outputs"}:
outputs.extend(split_csv(value))
elif key in {"fx", "effect", "effects"}:
effects.extend(split_csv(value))
elif key in {"no", "constraint", "constraints"}:
constraints.extend(split_csv(value))
elif key in {"alg", "algo", "algorithm"}:
algorithms.extend(split_csv(value))
elif key in {"tag", "tags"}:
tags.extend(split_csv(value))
elif key in {"rel", "relation", "relations", "uses", "partof", "before", "after"}:
relations.extend(f"{key}:{item}" for item in split_csv(value))
elif key in {"id", "intent_id"}:
intent_id = value
else:
# Nieznany key:value traktujemy jako tag pomocniczy.
tags.append(f"{key}:{value}")
continue
if ":" in token and not action and not object_name:
left, right = token.split(":", 1)
action = left
object_name = right
continue
positional.append(token)
if not action and positional:
action = positional[0]
if not object_name and len(positional) > 1:
object_name = "_".join(positional[1:])
if not action or not object_name:
return None
if scope not in VALID_SCOPES:
scope = default_scope
return IntentDescriptor(
action=action,
object=object_name,
priority=priority,
scope=scope,
domain=domain,
inputs=tuple(inputs),
outputs=tuple(outputs),
effects=tuple(effects),
constraints=tuple(constraints),
algorithms=tuple(algorithms),
tags=tuple(tags),
relations=tuple(relations),
intent_id=intent_id,
raw=payload,
)
def extract_intent_records_from_text(
source: str,
*,
file_path: str,
default_scope: str = "block",
) -> list[IntentRecord]:
"""
Wyciąga tagi @intent z dowolnego pliku tekstowego.
Działa dla:
- pojedynczej linii,
- funkcji,
- klasy,
- pliku,
- projektu, jeśli projektowy tag znajduje się np. w README/SUMD.
"""
records: list[IntentRecord] = []
for index, line in enumerate(source.splitlines(), start=1):
descriptor = parse_intent_tag_line(line, default_scope=default_scope)
if descriptor is None:
continue
records.append(
IntentRecord(
descriptor=descriptor,
file_path=file_path,
start_line=index,
end_line=index,
owner="",
)
)
return records
def normalize_label(value: str) -> str:
"""
Normalizuje nazwy techniczne:
user_permissions -> user_permission
UserPermission -> user_permission
scan-config -> scan_config
"""
value = value.strip()
value = re.sub(r"(?<=[a-z])(?=[A-Z])", "_", value)
value = value.replace("-", "_").replace("/", "_").replace(".", "_")
value = re.sub(r"[^A-Za-z0-9_]+", "_", value)
value = re.sub(r"_+", "_", value)
value = value.strip("_").lower()
parts = []
for part in value.split("_"):
if not part or part in STOP_WORDS:
continue
# Prosta liczba mnoga: permissions -> permission, extensions -> extension
if len(part) > 4 and part.endswith("s"):
part = part[:-1]
parts.append(part)
return "_".join(parts)
def normalize_action(action: str) -> str:
normalized = normalize_label(action)
for canonical, variants in ACTION_SYNONYMS.items():
normalized_variants = {normalize_label(item) for item in variants}
if normalized in normalized_variants:
return canonical
return normalized
def normalize_many(values: Iterable[str]) -> tuple[str, ...]:
normalized = {normalize_label(value) for value in values}
normalized.discard("")
return tuple(sorted(normalized))
def make_block_id(file_path: str, start_line: int, end_line: int, scope: str) -> str:
raw = f"{file_path}:{start_line}:{end_line}:{scope}"
return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
def build_intent_signature(record: IntentRecord) -> IntentSignature:
descriptor = record.descriptor
action = normalize_action(descriptor.action)
object_name = normalize_label(descriptor.object)
domain = normalize_label(descriptor.domain)
inputs = normalize_many(descriptor.inputs)
outputs = normalize_many(descriptor.outputs)
effects = normalize_many(descriptor.effects)
constraints = normalize_many(descriptor.constraints)
algorithms = normalize_many(descriptor.algorithms)
tags = normalize_many(descriptor.tags)
relations = normalize_many(descriptor.relations)
features: set[str] = {
f"action:{action}",
f"object:{object_name}",
f"priority:{descriptor.priority}",
f"scope:{descriptor.scope}",
}
if domain:
features.add(f"domain:{domain}")
features.update(f"in:{item}" for item in inputs)
features.update(f"out:{item}" for item in outputs)
features.update(f"fx:{item}" for item in effects)
features.update(f"no:{item}" for item in constraints)
features.update(f"alg:{item}" for item in algorithms)
features.update(f"tag:{item}" for item in tags)
features.update(f"rel:{item}" for item in relations)
exact_hash = hashlib.sha256(
"\n".join(sorted(features)).encode("utf-8")
).hexdigest()
block_id = descriptor.intent_id or make_block_id(
record.file_path,
record.start_line,
record.end_line,
descriptor.scope,
)
return IntentSignature(
block_id=block_id,
file_path=record.file_path,
start_line=record.start_line,
end_line=record.end_line,
scope=descriptor.scope,
action=action,
object=object_name,
domain=domain,
priority=descriptor.priority,
features=frozenset(features),
exact_hash=exact_hash,
raw=descriptor.raw,
)
def feature_values(signature: IntentSignature, prefix: str) -> set[str]:
marker = f"{prefix}:"
return {
feature[len(marker):]
for feature in signature.features
if feature.startswith(marker)
}
def jaccard(left: set[str], right: set[str]) -> float:
if not left and not right:
return 1.0
if not left or not right:
return 0.0
return len(left & right) / len(left | right)
def object_similarity(left: str, right: str) -> float:
if left == right:
return 1.0
left_parts = set(left.split("_"))
right_parts = set(right.split("_"))
return jaccard(left_parts, right_parts)
def priority_similarity(left: int, right: int) -> float:
return max(0.0, 1.0 - abs(left - right) / 4.0)
def score_intent_similarity(
left: IntentSignature,
right: IntentSignature,
) -> tuple[float, dict[str, object]]:
"""
Liczy podobieństwo dwóch intencji.
Scope nie blokuje porównania. Dzięki temu można porównać np.
funkcję z klasą albo plik z modułem.
"""
action_score = 1.0 if left.action == right.action else 0.0
object_score = object_similarity(left.object, right.object)
if left.domain or right.domain:
domain_score = 1.0 if left.domain == right.domain else 0.0
else:
domain_score = 1.0
input_score = jaccard(feature_values(left, "in"), feature_values(right, "in"))
output_score = jaccard(feature_values(left, "out"), feature_values(right, "out"))
effect_score = jaccard(feature_values(left, "fx"), feature_values(right, "fx"))
constraint_score = jaccard(feature_values(left, "no"), feature_values(right, "no"))
algorithm_score = jaccard(feature_values(left, "alg"), feature_values(right, "alg"))
tag_score = jaccard(feature_values(left, "tag"), feature_values(right, "tag"))
priority_score = priority_similarity(left.priority, right.priority)
# Scope jest małą wagą, bo różne poziomy mogą mieć tę samą intencję.
scope_score = 1.0 if left.scope == right.scope else 0.5
score = (
0.24 * action_score
+ 0.22 * object_score
+ 0.10 * domain_score
+ 0.10 * input_score
+ 0.10 * output_score
+ 0.06 * effect_score
+ 0.05 * constraint_score
+ 0.04 * algorithm_score
+ 0.04 * tag_score
+ 0.03 * priority_score
+ 0.02 * scope_score
)
reason: dict[str, object] = {
"action_score": round(action_score, 3),
"object_score": round(object_score, 3),
"domain_score": round(domain_score, 3),
"input_score": round(input_score, 3),
"output_score": round(output_score, 3),
"effect_score": round(effect_score, 3),
"constraint_score": round(constraint_score, 3),
"algorithm_score": round(algorithm_score, 3),
"tag_score": round(tag_score, 3),
"priority_score": round(priority_score, 3),
"scope_score": round(scope_score, 3),
"left_scope": left.scope,
"right_scope": right.scope,
"left": f"{left.action}:{left.object}",
"right": f"{right.action}:{right.object}",
}
return round(score, 4), reason
def bucket_intent_signatures(
signatures: Iterable[IntentSignature],
) -> dict[str, list[IntentSignature]]:
"""
Grupuje kandydatów, żeby nie robić pełnego O(n²) na całym projekcie.
Najpierw porównujemy tylko intencje z tą samą akcją.
"""
buckets: dict[str, list[IntentSignature]] = defaultdict(list)
for signature in signatures:
buckets[signature.action].append(signature)
return buckets
def find_intent_duplicate_pairs(
signatures: list[IntentSignature],
*,
threshold: float = 0.84,
) -> list[IntentDuplicatePair]:
"""
Zwraca pary intencji uznane za duplikaty.
"""
pairs: list[IntentDuplicatePair] = []
buckets = bucket_intent_signatures(signatures)
for bucket in buckets.values():
for left, right in combinations(bucket, 2):
if left.block_id == right.block_id:
continue
if left.exact_hash == right.exact_hash:
score = 1.0
reason: dict[str, object] = {
"exact_intent_hash": True,
"left": f"{left.action}:{left.object}",
"right": f"{right.action}:{right.object}",
}
else:
score, reason = score_intent_similarity(left, right)
if score >= threshold:
pairs.append(
IntentDuplicatePair(
left_id=left.block_id,
right_id=right.block_id,
similarity=score,
reason=reason,
)
)
return pairs
def group_intent_duplicate_pairs(
pairs: list[IntentDuplicatePair],
) -> list[set[str]]:
"""
Grupuje pary w większe klastry:
A podobne do B
B podobne do C
=> A, B, C w jednej grupie.
"""
parent: dict[str, str] = {}
def find(node: str) -> str:
parent.setdefault(node, node)
if parent[node] != node:
parent[node] = find(parent[node])
return parent[node]
def union(left: str, right: str) -> None:
root_left = find(left)
root_right = find(right)
if root_left != root_right:
parent[root_right] = root_left
for pair in pairs:
union(pair.left_id, pair.right_id)
groups: dict[str, set[str]] = defaultdict(set)
for node in parent:
groups[find(node)].add(node)
return [group for group in groups.values() if len(group) > 1]
def extract_intent_signatures_from_text(
source: str,
*,
file_path: str,
default_scope: str = "block",
) -> list[IntentSignature]:
"""
Wygodna funkcja end-to-end:
tekst pliku -> rekordy intencji -> sygnatury.
"""
records = extract_intent_records_from_text(
source,
file_path=file_path,
default_scope=default_scope,
)
return [build_intent_signature(record) for record in records]
def find_intent_duplicates_in_sources(
sources: dict[str, str],
*,
threshold: float = 0.84,
default_scope: str = "block",
) -> tuple[list[IntentSignature], list[IntentDuplicatePair], list[set[str]]]:
"""
Analiza wielu plików naraz.
sources:
{
"src/a.py": "...",
"src/b.py": "..."
}
"""
signatures: list[IntentSignature] = []
for file_path, source in sources.items():
signatures.extend(
extract_intent_signatures_from_text(
source,
file_path=file_path,
default_scope=default_scope,
)
)
pairs = find_intent_duplicate_pairs(signatures, threshold=threshold)
groups = group_intent_duplicate_pairs(pairs)
return signatures, pairs, groupssource_a = """
# @intent parse:extensions !p2 @cli in:raw out:list fx:none #config scope:function
def _parse_extensions(raw):
return raw.split(",")
"""
source_b = """
# @intent read:extension_list !p2 @cli in:raw_extensions out:list fx:none #config scope:function
def parse_ext(value):
return [x.strip() for x in value.split(",")]
"""
signatures, pairs, groups = find_intent_duplicates_in_sources(
{
"src/a.py": source_a,
"src/b.py": source_b,
},
threshold=0.84,
)
print(pairs)
print(groups)Ten przykład powinien wykryć podobieństwo, bo:
read -> parse
extensions / extension_list -> podobny object
domain = cli
inputs podobne
outputs podobne
tag = config
# @intent normalize:path !p3 @utils in:raw_path out:path scope:line
path = Path(raw).resolve()# @intent filter:test_files !p3 @scanner in:file_paths out:source_files scope:block
files = [p for p in files if not is_test_file(p)]# @intent parse:extensions !p2 @cli in:raw_extensions out:extension_list scope:function
def _parse_extensions(raw):
...# @intent manage:hash_cache !p2 @cache in:file_hash out:cached_result fx:read,write scope:class
class HashCache:
...Na początku pliku:
# @intent expose:scan_cli !p2 @cli in:path,options out:DuplicationMap scope:fileW README.md, SUMD.md albo .redup.intent:
<!-- @intent analyze:code_duplication !p1 @project in:source_tree out:refactor_plan scope:project #dedup #analysis -->
Gdy będziesz podpinał to pod reDUP, możesz użyć takiego adaptera. Jest odporny na różne nazwy pól, bo nie wiem, jak dokładnie masz obecnie zdefiniowany CodeBlock.
def extract_intent_signatures_from_code_blocks(
blocks: list[object],
*,
threshold_scope: str = "function",
) -> list[IntentSignature]:
"""
Adapter pod istniejące CodeBlock z reDUP.
Oczekiwane pola, zależnie od obecnego modelu:
- file_path albo path
- start_line albo start
- end_line albo end
- content albo text albo source
"""
signatures: list[IntentSignature] = []
for block in blocks:
file_path = str(
getattr(block, "file_path", "")
or getattr(block, "path", "")
or "<unknown>"
)
start_line = int(
getattr(block, "start_line", 1)
or getattr(block, "start", 1)
or 1
)
end_line = int(
getattr(block, "end_line", start_line)
or getattr(block, "end", start_line)
or start_line
)
text = (
getattr(block, "content", "")
or getattr(block, "text", "")
or getattr(block, "source", "")
or ""
)
records = extract_intent_records_from_text(
text,
file_path=file_path,
default_scope=threshold_scope,
)
for record in records:
# Korekta linii względem położenia bloku w pliku.
adjusted = IntentRecord(
descriptor=record.descriptor,
file_path=file_path,
start_line=start_line + record.start_line - 1,
end_line=start_line + record.end_line - 1,
owner=getattr(block, "function_name", "") or getattr(block, "class_name", ""),
)
signatures.append(build_intent_signature(adjusted))
return signaturesTak, szczególnie dla przypadków, których zwykłe narzędzia nie łapią:
1. Kod wygląda inaczej, ale robi to samo.
2. Funkcje mają inne nazwy, ale tę samą odpowiedzialność.
3. Ten sam obowiązek jest rozlany między klasą, funkcją i helperem.
4. Projekt ma duplikację architektoniczną, nie tylko tekstową.
5. Chcesz wykrywać duplikaty między językami, np. Python/JS/C#.
Największa zaleta jest taka, że ten format:
# @intent build:scan_config !p1 @scanner in:path,params out:ScanConfig fx:none #configjest jednocześnie:
czytelny dla człowieka,
łatwy do parsowania regexem,
niezależny od języka programowania,
porównywalny algorytmicznie,
krótki,
skalowalny od linii do projektu.
Czyli tak: to może stać się Twoją warstwą intencyjnych fingerprintów kodu.