Python正则表达式在数据处理中的应用实战案例

import re

# 最简单的例子：在字符串中查找 'hello'
text = "world hello python"
pattern = "hello"

# re.search() 扫描整个字符串，返回第一个匹配的Match对象
match = re.search(pattern, text)
if match:
    print(f"Found '{match.group()}' at position {match.start()} to {match.end()}")
# 输出: Found 'hello' at position 6 to 11

import re

def validate_username(username):
    pattern = r'^[a-zA-Z]\w{3,15}$'
    # ^[a-zA-Z]   : 必须以字母开头
    # \w{3,15}    : 后面跟3到15个单词字符（总长度就是4-16）
    # $           : 必须以此结尾
    if re.fullmatch(pattern, username): # fullmatch确保整个字符串匹配模式
        return True
    else:
        return False

print(validate_username("Alice123"))    # True
print(validate_username("alice"))       # True (长度4)
print(validate_username("23alice"))     # False (数字开头)
print(validate_username("a"))           # False (太短)
print(validate_username("alice!#"))     # False (包含非法字符)

text = "The price is 100 dollars"

print(re.match(r'\d+', text))    # None (开头不是数字)
print(re.search(r'\d+', text))   # <re.Match object; span=(12, 15), match='100'>

text = "Apple: $1.99, Banana: $0.50, Orange: $2.00"

# 提取所有价格
prices = re.findall(r'\$\d+\.\d\d', text) # \.$ 转义，匹配真正的美元符号
print(prices)  # ['$1.99', '$0.50', '$2.00']

# 使用finditer获取更多信息（如位置）
for match in re.finditer(r'\$\d+\.\d\d', text):
    print(f"Found {match.group()} at {match.span()}")
# Found $1.99 at (7, 12)
# Found $0.50 at (23, 28)
# Found $2.00 at (38, 43)

text = "Today is 2023-10-27. The meeting is on 2023-11-01."

# 将日期格式从 YYYY-MM-DD 改为 DD/MM/YYYY
# 使用分组 \1, \2, \3 来引用前面匹配的内容
new_text = re.sub(r'(\d{4})-(\d{2})-(\d{2})', r'\3/\2/\1', text)
print(new_text) # Today is 27/10/2023. The meeting is on 01/11/2023.

# 使用函数进行更复杂的替换（将所有年份加1）
def add_one_year(match_obj):
    year = int(match_obj.group(1))
    month = match_obj.group(2)
    day = match_obj.group(3)
    return f"{day}/{month}/{year + 1}" # 返回要替换成的字符串

new_text_func = re.sub(r'(\d{4})-(\d{2})-(\d{2})', add_one_year, text)
print(new_text_func) # Today is 27/10/2024. The meeting is on 01/11/2024.

import re

# 未编译模式，每次调用都要解释一遍模式
result1 = re.findall(r'\d+', text1)
result2 = re.findall(r'\d+', text2)

# 编译模式，只需解释一次
pattern_obj = re.compile(r'\d+') # 返回一个Pattern对象
result1 = pattern_obj.findall(text1)
result2 = pattern_obj.findall(text2)

# 编译后的对象拥有所有re模块的方法：search, match, findall, sub等

text = "https://www.example.com http://blog.example.org"

# 只想捕获域名，不需要协议这个分组
# 使用 (?:...) 表示非捕获分组
domains = re.findall(r'(?:https?://)(\w+\.\w+\.\w+)', text)
print(domains) # ['www.example.com', 'blog.example.org']

# 对比捕获分组，结果会包含协议
domains_with_protocol = re.findall(r'(https?://)(\w+\.\w+\.\w+)', text)
print(domains_with_protocol) # [('https://', 'www.example.com'), ('http://', 'blog.example.org')]

html_text = "<title>Python Regular Expressions</title> <div>Content</div>"

# 贪婪匹配：.* 会匹配到最后一个 > 之前的所有字符
greedy_match = re.search(r'<.*>', html_text)
print(greedy_match.group()) # <title>Python Regular Expressions</title> <div>Content</div>

# 非贪婪匹配：.*? 在遇到第一个 > 后就停止匹配
non_greedy_match = re.search(r'<.*?>', html_text)
print(non_greedy_match.group()) # <title>

# 提取后面跟着 " dollars" 的数字
text = "100 dollars, 200 euros, 300 dollars"
dollars = re.findall(r'\d+(?=\s*dollars)', text) # 只匹配数字，不匹配后面的“dollars”
print(dollars) # ['100', '300']

# 提取前面是 "$" 的数字
text = "The prices are $100, €200, and $300."
prices = re.findall(r'(?<=\$)\d+', text) # 只匹配数字，不匹配前面的“$”
print(prices) # ['100', '300']

# 匹配一个不在单词中间的连字符（即单词边界处的连字符）
text = "multi-purpose and well-known, but not abc-123"
hyphens = re.findall(r'(?<=\w)-(?=\w)', text) # 前面是单词字符，后面也是单词字符的连字符
print(hyphens) # ['-', '-']  (匹配到 purpose和known中的连字符，但没匹配abc-123，因为123不是单词字符\w)

# nginx_log_analyzer.py
import re
from collections import Counter, defaultdict
from pathlib import Path

def analyze_nginx_log(log_path):
    """
    分析Nginx访问日志
    """
    # 使用编译正则表达式提高效率
    # 分解日志行的正则模式
    log_pattern = re.compile(
        r'(?P<ip>\d+\.\d+\.\d+\.\d+)'          # IP地址
        r' - - '                               # 忽略的标识
        r'\[(?P<time>.*?)\]'                   # 时间戳
        r' "(?P<request>.*?)"'                 # 请求行 (GET /url HTTP/1.1)
        r' (?P<status>\d{3})'                  # 状态码
        r' (?P<size>\d+)'                      # 响应大小
        r' "(?P<referer>.*?)"'                 # 来源页
        r' "(?P<agent>.*?)"'                   # User-Agent
    )

    # 用于统计的数据结构
    ip_counter = Counter()          # IP访问次数 (PV)
    status_counter = Counter()      # 状态码出现次数
    url_counter = Counter()         # 请求URL出现次数
    # 使用 defaultdict(set) 来记录每个IP对应的唯一User-Agent
    ip_ua_dict = defaultdict(set)   # key: ip, value: set(user_agent1, user_agent2...)

    try:
        with open(log_path, 'r', encoding='utf-8') as f:
            for line in f:
                match = log_pattern.search(line)
                if not match:
                    # 跳过无法识别的行
                    continue

                # 从匹配结果中提取数据
                data = match.groupdict()
                ip = data['ip']
                status = data['status']
                request = data['request']
                user_agent = data['agent']

                # 1. 统计IP PV
                ip_counter[ip] += 1

                # 2. 记录IP和User-Agent（用于估算UV）
                ip_ua_dict[ip].add(user_agent) # 使用set自动去重

                # 3. 统计状态码
                status_counter[status] += 1

                # 4. 从请求行中提取URL（简单处理，提取第一个单词后的部分）
                parts = request.split()
                if len(parts) >= 2:
                    url = parts[1] # 例如 "/articles/python.html"
                    url_counter[url] += 1

    except FileNotFoundError:
        print(f"Error: File {log_path} not found.")
        return

    # --- 输出分析报告 ---
    print("=" * 50)
    print("Nginx Log Analysis Report")
    print("=" * 50)

    print(f"\n1. Top 10 IP Addresses by Page Views (PV):")
    for ip, count in ip_counter.most_common(10):
        print(f"   {ip:15} : {count:4}")

    print(f"\n2. Estimated Unique Visitors (UV) per IP (by unique User-Agent):")
    # 计算每个IP的UV（即其唯一User-Agent集合的大小）
    ip_uv = {ip: len(ua_set) for ip, ua_set in ip_ua_dict.items()}
    for ip, uv_count in sorted(ip_uv.items(), key=lambda x: x[1], reverse=True)[:10]:
        print(f"   {ip:15} : {uv_count:4} (PV: {ip_counter[ip]})")

    print(f"\n3. HTTP Status Code Distribution:")
    for status, count in status_counter.most_common():
        print(f"   {status} : {count}")

    print(f"\n4. Top 10 Most Frequently Requested URLs:")
    for url, count in url_counter.most_common(10):
        print(f"   {count:4} : {url}")

if __name__ == "__main__":
    log_file = "access.log"  # 替换为你的日志文件路径
    analyze_nginx_log(log_file)