change-review

Change Review

Review of CRM and PM data changes before PR -- a data equivalent of code-review

Public repository: https://github.com/your-org/claude-change-review-skill

When to use

• Before a PR with CRM changes
• Before a PR with PM changes (tasks, projects)
• "check my CRM changes"
• "change review"
• "data review"

Paths

What	Path
CRM	$CRM_PATH/
CRM Schema	$CRM_PATH/schema.yaml
PM	$PM_PATH/

How to execute

Step 1: Read schema.yaml

# ALWAYS read the schema first -- it contains all the rules
cat $CRM_PATH/schema.yaml

Schema contains:

• primary_key -- unique identifier
• required -- required fields
• unique -- unique fields
• foreign_keys -- relationships between tables
• composite_unique -- composite unique keys
• enums -- allowed values
• id_format -- regex for ID format
• rules -- business rules

Step 2: Get the diff

git diff HEAD -- sales/crm/

Step 3: Load data

import pandas as pd
import yaml

# Read schema
with open('$CRM_PATH/schema.yaml') as f:
    schema = yaml.safe_load(f)

# Load all tables
base_path = '$CRM_PATH/'
tables = {}
for table_name, table_def in schema['tables'].items():
    tables[table_name] = pd.read_csv(base_path + table_def['file'])

Step 4: Validate against schema

import re

def validate_table(name, df, table_schema, all_tables):
    issues = []

    # 1. Primary key uniqueness
    pk = table_schema['primary_key']
    if df[pk].duplicated().any():
        dups = df[df[pk].duplicated()][pk].tolist()
        issues.append(('critical', f'Duplicate {pk}: {dups}'))

    # 2. Required fields
    for field in table_schema.get('required', []):
        missing = df[df[field].isna()]
        if len(missing) > 0:
            issues.append(('critical', f'Missing required {field}: {len(missing)} rows'))

    # 3. Unique fields
    for field in table_schema.get('unique', []):
        if field == pk:
            continue
        dups = df[df[field].notna() & df[field].duplicated()]
        if len(dups) > 0:
            issues.append(('high', f'Duplicate {field}: {dups[field].tolist()}'))

    # 4. Foreign keys
    for fk_field, ref in table_schema.get('foreign_keys', {}).items():
        ref_table, ref_field = ref.split('.')
        valid_values = all_tables[ref_table][ref_field]
        invalid = df[df[fk_field].notna() & ~df[fk_field].isin(valid_values)]
        if len(invalid) > 0:
            issues.append(('critical', f'Invalid {fk_field} references: {invalid[fk_field].tolist()}'))

    # 5. Composite unique
    for fields in table_schema.get('composite_unique', []):
        dups = df[df.duplicated(subset=fields, keep=False)]
        if len(dups) > 0:
            issues.append(('critical', f'Duplicate {"+".join(fields)} combination'))

    # 6. Enum values
    for field, valid_values in table_schema.get('enums', {}).items():
        invalid = df[df[field].notna() & ~df[field].isin(valid_values)]
        if len(invalid) > 0:
            issues.append(('high', f'Invalid {field} values: {invalid[field].unique().tolist()}'))

    # 7. ID format
    if 'id_format' in table_schema:
        pattern = table_schema['id_format']
        invalid = df[~df[pk].str.match(pattern, na=False)]
        if len(invalid) > 0:
            issues.append(('low', f'Invalid ID format: {invalid[pk].tolist()}'))

    return issues

# Validate all tables
for name, table_schema in schema['tables'].items():
    if name in tables:
        issues = validate_table(name, tables[name], table_schema, tables)
        for severity, msg in issues:
            print(f'[{severity.upper()}] {name}: {msg}')