仅解析在线文本文件以获取最新数据

Question

8/23 编辑：

感谢大家的回复以及可能比我的效率更高的代码。但是，我没有准确描述我想要捕捉的内容。

@DarkKnight 是正确的，我要查询的重要标记位于第 5 列。但是对于每个重要标记，我最多需要解析 15 行文本才能捕获完整的模型运行。以“TVCN”为例，我需要捕捉所有这些：

AL, 07, 2021082118, 03, TVCN, 0, 197N, 995W, 0

AL, 07, 2021082118, 03, TVCN, 12, 194N, 1026W, 0

AL, 07, 2021082118, 03, TVCN, 24, 191N, 1055W, 0

AL, 07, 2021082118, 03, TVCN, 36, 198N, 1084W, 0

AL, 07, 2021082118, 03, TVCN, 48, 202N, 1113W, 0

AL, 07, 2021082118, 03, TVCN, 60, 204N, 1139W, 0

AL, 07, 2021082118, 03, TVCN, 72, 208N, 1164W, 0

AL, 07, 2021082118, 03, TVCN, 84, 210N, 1188W, 0

AL, 07, 2021082118, 03, TVCN, 96, 211N, 1209W, 0

AL, 07, 2021082118, 03, TVCN, 108, 206N, 1230W, 0

AL, 07, 2021082118, 03, TVCN, 120, 201N, 1251W, 0

第 3 列是模型运行的日期/时间 (yyyymmddhh)，而第 6 列是预测时间。因此，为了通过时间绘制预测但仅捕获最近的模型运行，我需要返回所有日期为“2021082118”的 TVCN 实例。当然，每次模型再次运行时，日期值都会更新。这有意义吗？

---

我有部分代码可以满足我的需求，但我一直在努力将它准确地送到我想要的地方。我正在从在线文本文件中提取逗号分隔的数据。然后我的代码会抛出我不想要的行。这些是飓风预报模型的原始数据。但是，在线文本文件存储了给定风暴的所有模型运​​行。我只想提取我选择的模型的最新运行。对于给定的模型运行（预测 t+12、t+24 等），每个模型都有多行文本。这可以实现吗？

这是我所拥有的部分工作：

import urllib.request

webf = urllib.request.urlopen("http://hurricanes.ral.ucar.edu/realtime/plots/northatlantic/2021/al072021/aal072021.dat")
lines = webf.readlines() 

important_codes = ["AEM2", "AEMI", "AVNI", "CEM2", "COT2", "CTC1", "DSHP", "EGR2", "HMON", "HWFI", "NNIB", "LGEM", "NNIC", "OFCI", "OFCL", "SHIP", "TVCN", "UKX2"]

def is_important_line(line):
    return any(code in line for code in important_codes)

output_lines = []
for line in lines:
    decoded_line = line.decode("utf-8") 

    if not is_important_line(decoded_line):
        continue
    output_lines.append(decoded_line)

f = open('test.txt', 'w') 

f.write("".join(output_lines)) 
f.close()

Answer 1

好的，我过滤了错误的列。这应该工作

output_lines = []
for line in lines:
    line = line.decode("utf-8") 
    line = line.split(', ')[:-1]

    if line[4] not in important_codes:
        continue
    output_lines.append(line)
    
output_lines = sorted(output_lines, key=lambda x: x[4])
new_output = []
for code, group in groupby(output_lines, key=lambda x: x[4]):
    best_date = 0
    temp_entries = []
    for date, entries in groupby(group, key=lambda x: x[2]):
        date = int(date)
        if date > best_date:
            best_date = date
            temp_entries = list(entries)
    for entry in temp_entries:
        new_output.append(', '.join(entry))

with open('mydata.dat', 'w') as f:
    f.write('\n'.join(new_output))

Answer 2

如果在迭代输入数据时编写输出文件可能会更好。 “重要”标记似乎在第 5 列（基数 1）中。例如，如果“AVNI”出现在一行中的其他位置，您的代码可能会导致模棱两可的结果。试试这个：-

import requests

IC = ["AEM2", "AEMI", "AVNI", "CEM2", "COT2", "CTC1", "DSHP", "EGR2", "HMON",
      "HWFI", "NNIB", "LGEM", "NNIC", "OFCI", "OFCL", "SHIP", "TVCN", "UKX2"]

with open('test.txt', 'w') as outfile:
    with requests.Session() as session:
        response = session.get(
            'http://hurricanes.ral.ucar.edu/realtime/plots/northatlantic/2021/al072021/aal072021.dat')
        response.raise_for_status()
        for line in response.text.splitlines():
            try:
                if line.split(',')[4].strip() in IC:
                    outfile.write(f'{line}\n')
            except IndexError:
                pass
print('Done')

编辑：如果您只对最近出现的“重要”令牌感兴趣，那么您可以这样做：-

import requests

IC = ["AEM2", "AEMI", "AVNI", "CEM2", "COT2", "CTC1", "DSHP", "EGR2", "HMON",
      "HWFI", "NNIB", "LGEM", "NNIC", "OFCI", "OFCL", "SHIP", "TVCN", "UKX2"]

with open('test.txt', 'w') as outfile:
    TD = {}
    with requests.Session() as session:
        response = session.get(
            'http://hurricanes.ral.ucar.edu/realtime/plots/northatlantic/2021/al072021/aal072021.dat')
        response.raise_for_status()
        for line in response.text.splitlines():
            try:
                if (k := line.split(',')[4].strip()) in IC:
                    TD[k] = line
            except IndexError:
                pass
    for v in TD.values():
        outfile.write(f'{v}\n')
print('Done')