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)