14：全球犯罪记录数据库构建：户籍+公开档案的SQL/NoSQL整合架构

-- 人员表
CREATE TABLE person (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    id_number VARCHAR(20) UNIQUE NOT NULL,
    date_of_birth DATE,
    gender VARCHAR(10),
    nationality VARCHAR(50),
    address VARCHAR(200),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 犯罪记录表
CREATE TABLE crime_record (
    id SERIAL PRIMARY KEY,
    person_id INTEGER REFERENCES person(id),
    crime_type VARCHAR(100) NOT NULL,
    crime_date DATE NOT NULL,
    crime_location VARCHAR(200),
    description TEXT,
    severity INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 索引
CREATE INDEX idx_person_name ON person(name);
CREATE INDEX idx_person_id_number ON person(id_number);
CREATE INDEX idx_crime_record_person_id ON crime_record(person_id);
CREATE INDEX idx_crime_record_crime_type ON crime_record(crime_type);

// 公开档案文档结构
{
  "person_id": 123,
  "documents": [
    {
      "type": "arrest_record",
      "content": "...",
      "date": "2025-01-01",
      "source": "police_report"
    },
    {
      "type": "court_record",
      "content": "...",
      "date": "2025-02-01",
      "source": "court_database"
    }
  ],
  "social_media": {
    "twitter": "@username",
    "facebook": "facebook.com/username",
    "instagram": "@username"
  },
  "images": [
    {
      "url": "https://example.com/image1.jpg",
      "date": "2025-01-01",
      "source": "surveillance"
    }
  ]
}

import pika
import json
import psycopg2
from pymongo import MongoClient

# 连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='data_sync')

# 连接到PostgreSQL
pg_conn = psycopg2.connect(
    host="localhost",
    database="crime_db",
    user="admin",
    password="password"
)

# 连接到MongoDB
mongo_client = MongoClient("mongodb://localhost:27017/")
mongo_db = mongo_client["crime_db"]
mongo_collection = mongo_db["person_documents"]

def callback(ch, method, properties, body):
    data = json.loads(body)
    if data["type"] == "person_update":
        # 更新PostgreSQL中的人员信息
        cursor = pg_conn.cursor()
        cursor.execute(
            "UPDATE person SET name = %s, address = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s",
            (data["name"], data["address"], data["id"])
        )
        pg_conn.commit()
        cursor.close()
        
        # 更新MongoDB中的相关文档
        mongo_collection.update_one(
            {"person_id": data["id"]},
            {"$set": {"last_updated": data["updated_at"]}}
        )
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='data_sync', on_message_callback=callback)
print("Waiting for messages. To exit press CTRL+C")
channel.start_consuming()

from elasticsearch import Elasticsearch

# 连接到Elasticsearch
es = Elasticsearch(["localhost:9200"])

def index_person(person_id, name, crime_records):
    # 构建索引文档
    doc = {
        "person_id": person_id,
        "name": name,
        "crime_records": crime_records,
        "last_updated": "2025-03-15T00:00:00"
    }
    
    # 索引文档
    es.index(
        index="crime_records",
        id=person_id,
        body=doc
    )

def search_person(name, crime_type=None):
    # 构建搜索查询
    query = {
        "query": {
            "bool": {
                "must": [
                    {"match": {"name": name}}
                ]
            }
        }
    }
    
    if crime_type:
        query["query"]["bool"]["must"].append({"match": {"crime_records.type": crime_type}})
    
    # 执行搜索
    result = es.search(index="crime_records", body=query)
    return result["hits"]["hits"]

import requests
from bs4 import BeautifulSoup
import pandas as pd

def scrape_court_records(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')
    
    # 提取法院记录
    records = []
    for record in soup.find_all('div', class_='court-record'):
        name = record.find('span', class_='name').text
        crime_type = record.find('span', class_='crime-type').text
        date = record.find('span', class_='date').text
        
        records.append({
            'name': name,
            'crime_type': crime_type,
            'date': date
        })
    
    return pd.DataFrame(records)

# MongoDB分片配置
# 在MongoDB shell中执行
# sh.enableSharding("crime_db")
# sh.shardCollection("crime_db.person_documents", {"person_id": 1})

# PostgreSQL分区表
# CREATE TABLE crime_record_partitioned (
#     id SERIAL,
#     person_id INTEGER,
#     crime_type VARCHAR(100),
#     crime_date DATE,
#     PRIMARY KEY (id, crime_date)
# ) PARTITION BY RANGE (crime_date);
# 
# CREATE TABLE crime_record_2024 PARTITION OF crime_record_partitioned
#     FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');
# 
# CREATE TABLE crime_record_2025 PARTITION OF crime_record_partitioned
#     FOR VALUES FROM ('2025-01-01') TO ('2026-01-01');

# Redis缓存示例
import redis

# 连接到Redis
r = redis.Redis(host='localhost', port=6379, db=0)

def get_person_info(person_id):
    # 尝试从缓存获取
    cache_key = f"person:{person_id}"
    cached_data = r.get(cache_key)
    
    if cached_data:
        return json.loads(cached_data)
    
    # 从数据库获取
    cursor = pg_conn.cursor()
    cursor.execute("SELECT * FROM person WHERE id = %s", (person_id,))
    person = cursor.fetchone()
    cursor.close()
    
    if person:
        # 存入缓存
        person_dict = {
            'id': person[0],
            'name': person[1],
            'id_number': person[2],
            'date_of_birth': str(person[3]),
            'gender': person[4],
            'nationality': person[5],
            'address': person[6]
        }
        r.setex(cache_key, 3600, json.dumps(person_dict))
        return person_dict
    
    return None