Request atarken Python'da "SSL: Certıfıcate_verıfy_faıled" hatası

tugozden

Femtopat
Katılım
18 Mart 2025
Mesajlar
13
Daha fazla  
Cinsiyet
Erkek
Merhaba, Python'da rasathane ve usgsden API ile veri çekip bunu DC botuyla servera atan bir kod yazdım ama sürekli bu hatayı alıyorum. Ne yapacağım? (Kod biraz uzun ama gerekirse atarım.)

 
Son düzenleyen: Moderatör:
  • Rate limit (API sınırı) aşılıyor olabilir.
  • Yanıt formatı değişmiş olabilir (örneğin JSON beklerken başka bir şey gelmiş).
  • Zamanlama/senkronizasyon sorunları olabilir (örneğin async fonksiyonlar yanlış kullanılmış).
  • API key eksik ya da yanlış olabilir.
  • Bot yetkileri eksik olabilir (özellikle mesaj gönderme veya embed atma yetkisi).
 
TheAny dedi:
Kodu paylasir misin? Discord'un sertifikalarini self signed sertifika deyip reddetmis gibi duruyor.
Genişletmek için tıkla...

Python: 
import requests.
from datetime import datetime, timezone.
import discord.
from discord.ext import tasks.
import certifi.
import os.
import ssl.
import logging.
os.environ['DISCORD_PY_NO_VERIFY_SSL'] = '1'.
os.environ['SSL_CERT_FILE'] = certifi.where()

# Log ayarları.
logging.basicConfig(
 level=logging.INFO,
 format='%(asctime)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler('deprem_bot.log'),
 logging.StreamHandler()
 ]
)

# SSL sertifika ayarı.
os.environ["SSL_CERT_FILE"] = certifi.where()

# Ayarlar.
MIN_MAGNITUDE = 4.0 # Bildirim yapılacak minimum büyüklük.
CHECK_INTERVAL = 300 # Kontrol aralığı (saniye)
DISCORD_TOKEN = 'Bot_tokeni' # Discord bot tokeniniz.
DISCORD_CHANNEL_ID = 12345 # Hedef kanal ID.

# Son kontrol zamanı.
last_checked_time = None.

# Discord client ayarları.
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())
intents = discord.Intents.default()
intents.message_content = True.

class DepremBot(discord.Client):
 def __init__(self, *args, **kwargs):
 super().__init__(*args, **kwargs)
 self.logger = logging.getLogger(__name__)
 self.check_task = None.

 async def setup_hook(self):
 self.check_task = self.loop.create_task(self.periodic_check())

 async def periodic_check(self):
 await self.wait_until_ready()
 while not self.is_closed():
 try:
 await self.check_earthquakes()
 except Exception as e:
 self.logger.error(f"Kontrol sırasında hata: {e}")
 await asyncio.sleep(CHECK_INTERVAL)

 async def check_earthquakes(self):
 global last_checked_time.

 self.logger.info("Deprem verileri kontrol ediliyor...")

 # Verileri al.
 earthquakes = await self.get_combined_data()

 if not earthquakes:
 self.logger.warning("Hiç deprem verisi alınamadı")
 return.

 # Yeni depremleri filtrele.
 new_earthquakes = [
 eq for eq in earthquakes.
 if eq['magnitude'] >= MIN_MAGNITUDE and.
 (last_checked_time is None or eq['time'] > last_checked_time)
 ]

 if not new_earthquakes:
 self.logger.info("Yeni deprem bulunamadı")
 return.

 # Zamanı güncelle.
 last_checked_time = max(eq['time'] for eq in earthquakes)

 # Bildirim gönder.
 channel = self.get_channel(DISCORD_CHANNEL_ID)
 if channel is None:
 self.logger.error("Kanal bulunamadı!")
 return.

 for eq in sorted(new_earthquakes, key=lambda x: x['time'], reverse=True):
 try:
 embed = self.create_embed(eq)
 await channel.send(embed=embed)
 self.logger.info(f"Bildirim gönderildi: {eq['magnitude']}M - {eq['location']}")
 except Exception as e:
 self.logger.error(f"Bildirim gönderilemedi: {e}")

 async def get_combined_data(self):
 """Tüm kaynaklardan verileri birleştirir"""
 results = []

 # Rasathane verileri.
 try:
 rasathane_url = "https://www.mertsenturk.net/deprem/api"
 response = await self.loop.run_in_executor(None, lambda: requests.get(rasathane_url, timeout=10))
 response.raise_for_status()

 for eq in response.json():
 try:
 results.append({
 'magnitude': float(eq['buyukluk']),
 'location': eq['yer'],
 'time': datetime.fromisoformat(eq['tarih'].replace('Z', '+00:00')),
 'depth': eq['derinlik'],
 'source': 'Rasathane'.
 })
 except (KeyError, ValueError) as e:
 self.logger.warning(f"Rasathane veri hatası: {e}")

 except Exception as e:
 self.logger.error(f"Rasathane bağlantı hatası: {e}")

 # USGS verileri.
 try:
 usgs_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_day.geojson"
 response = await self.loop.run_in_executor(None, lambda: requests.get(usgs_url, timeout=10))
 response.raise_for_status()
 data = response.json()

 for feature in data['features']:
 try:
 props = feature['properties']
 results.append({
 'magnitude': props['mag'],
 'location': props['place'],
 'time': datetime.fromtimestamp(props['time']/1000, tz=timezone.utc),
 'depth': feature['geometry']['coordinates'][2],
 'source': 'USGS',
 'detail_url': props.get('url', '')
 })
 except (KeyError, ValueError) as e:
 self.logger.warning(f"USGS veri hatası: {e}")

 except Exception as e:
 self.logger.error(f"USGS bağlantı hatası: {e}")

 return results.

 def create_embed(self, earthquake):
 """Discord embed mesajı oluşturur"""
 # Renk büyüklüğe göre ayarla.
 if earthquake['magnitude'] >= 6.0:
 color = 0xff0000 # Kırmızı.
 elif earthquake['magnitude'] >= 5.0:
 color = 0xff6600 # Turuncu.
 else:
 color = 0xffcc00 # Sarı.

 embed = discord.Embed(
 title=f"🚨 Deprem - {earthquake['magnitude']:.1f} M",
 description=earthquake['location'],
 color=color,
 timestamp=earthquake['time']
 )

 # Alanlar.
 embed.add_field(name="📏 Büyüklük", value=f"{earthquake['magnitude']:.1f}", inline=True)
 embed.add_field(name="⬇️ Derinlik", value=f"{earthquake['depth']:.1f} km", inline=True)
 embed.add_field(name="🕒 Yerel Zaman",
 value=earthquake['time'].astimezone().strftime("%d.%m.%Y %H:%M:%S"),
 inline=False)
 embed.add_field(name="🔍 Kaynak", value=earthquake['source'], inline=True)

 # Detay linki (USGS için)
 if earthquake.get('detail_url'):
 embed.url = earthquake['detail_url']

 embed.set_footer(text="Deprem İzleme Botu • Veriler anlık değildir")
 return embed.

 async def on_ready(self):
 """Bot hazır olduğunda çalışır"""
 self.logger.info(f'Bot {self.user} olarak giriş yapıldı')
 self.logger.info(f'{len(self.guilds)} sunucuda aktif')

 # Bot durumunu ayarla.
 await self.change_presence(
 activity=discord.Activity(
 type=discord.ActivityType.watching,
 name=f"{MIN_MAGNITUDE}+ şiddetinde depremler"
 )
 )

 async def on_disconnect(self):
 """Bağlantı kesildiğinde çalışır"""
 global last_checked_time.
 if last_checked_time:
 try:
 with open('last_checked.txt', 'w') as f:
 f.write(last_checked_time.isoformat())
 self.logger.info("Son kontrol zamanı kaydedildi")
 except Exception as e:
 self.logger.error(f"Son kontrol zamanı kaydedilemedi: {e}")

# Başlangıç.
if __name__ == "__main__":
 import asyncio.

 # Son kontrol zamanını yükle.
 try:
 with open('last_checked.txt', 'r') as f:
 last_checked_time = datetime.fromisoformat(f.read().strip())
 logging.info(f"Son kontrol: {last_checked_time}")
 except (FileNotFoundError, ValueError):
 last_checked_time = None.
 logging.warning("Son kontrol zamanı bulunamadı")

 # Botu başlat.
 bot = DepremBot(intents=intents, ssl=ssl_context)

 try:
 logging.info("Bot başlatılıyor...")
 bot.run(DISCORD_TOKEN)
 except discord.LoginFailure:
 logging.error("Hata: Geçersiz bot tokeni!")
 except KeyboardInterrupt:
 logging.info("Bot kapatılıyor...")
 except Exception as e:
 logging.error(f"Beklenmeyen hata: {type(e).__name__}: {e}")
 
Lutfen mesaji duzenleyip tekrar atin. Atarken bu sefer sunun secili olmadigindan emin olun ve duzelt tusuna basmadiginizdam emin olun;
 
import requests
from datetime import datetime, timezone
import discord
from discord.ext import tasks
import certifi
import os
import ssl
import logging
os.environ['DISCORD_PY_NO_VERIFY_SSL'] = '1'
os.environ['SSL_CERT_FILE'] = certifi.where()

# Log ayarları
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('deprem_bot.log'),
logging.StreamHandler()
]
)
# SSL sertifika ayarı
os.environ["SSL_CERT_FILE"] = certifi.where()
# Ayarlar
MIN_MAGNITUDE = 4.0 # Bildirim yapılacak minimum büyüklük
CHECK_INTERVAL = 300 # Kontrol aralığı (saniye)
DISCORD_TOKEN = 'Bot_tokeni' # Discord bot tokeniniz
DISCORD_CHANNEL_ID = 12345 # Hedef kanal ID
# Son kontrol zamanı
last_checked_time = None
# Discord client ayarları
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(certifi.where())
intents = discord.Intents.default()
intents.message_content = True
class DepremBot(discord.Client):
def init(self, *args, **kwargs):
super().init(*args, **kwargs)
self.logger = logging.getLogger(name)
self.check_task = None
async def setup_hook(self):
self.check_task = self.loop.create_task(self.periodic_check())
async def periodic_check(self):
await self.wait_until_ready()
while not self.is_closed():
try:
await self.check_earthquakes()
except Exception as e:
self.logger.error(f"Kontrol sırasında hata: {e}")
await asyncio.sleep(CHECK_INTERVAL)
async def check_earthquakes(self):
global last_checked_time

self.logger.info("Deprem verileri kontrol ediliyor...")

# Verileri al
earthquakes = await self.get_combined_data()

if not earthquakes:
self.logger.warning("Hiç deprem verisi alınamadı")
return

# Yeni depremleri filtrele
new_earthquakes = [
eq for eq in earthquakes
if eq['magnitude'] >= MIN_MAGNITUDE and
(last_checked_time is None or eq['time'] > last_checked_time)
]

if not new_earthquakes:
self.logger.info("Yeni deprem bulunamadı")
return

# Zamanı güncelle
last_checked_time = max(eq['time'] for eq in earthquakes)

# Bildirim gönder
channel = self.get_channel(DISCORD_CHANNEL_ID)
if channel is None:
self.logger.error("Kanal bulunamadı!")
return

for eq in sorted(new_earthquakes, key=lambda x: x['time'], reverse=True):
try:
embed = self.create_embed(eq)
await channel.send(embed=embed)
self.logger.info(f"Bildirim gönderildi: {eq['magnitude']}M - {eq['location']}")
except Exception as e:
self.logger.error(f"Bildirim gönderilemedi: {e}")
async def get_combined_data(self):
"""Tüm kaynaklardan verileri birleştirir"""
results = []

# Rasathane verileri
try:
rasathane_url = "https://www.mertsenturk.net/deprem/api"
response = await self.loop.run_in_executor(None, lambda: requests.get(rasathane_url, timeout=10))
response.raise_for_status()

for eq in response.json():
try:
results.append({
'magnitude': float(eq['buyukluk']),
'location': eq['yer'],
'time': datetime.fromisoformat(eq['tarih'].replace('Z', '+00:00')),
'depth': eq['derinlik'],
'source': 'Rasathane'
})
except (KeyError, ValueError) as e:
self.logger.warning(f"Rasathane veri hatası: {e}")

except Exception as e:
self.logger.error(f"Rasathane bağlantı hatası: {e}")
# USGS verileri
try:
usgs_url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_day.geojson"
response = await self.loop.run_in_executor(None, lambda: requests.get(usgs_url, timeout=10))
response.raise_for_status()
data = response.json()

for feature in data['features']:
try:
props = feature['properties']
results.append({
'magnitude': props['mag'],
'location': props['place'],
'time': datetime.fromtimestamp(props['time']/1000, tz=timezone.utc),
'depth': feature['geometry']['coordinates'][2],
'source': 'USGS',
'detail_url': props.get('url', '')
})
except (KeyError, ValueError) as e:
self.logger.warning(f"USGS veri hatası: {e}")

except Exception as e:
self.logger.error(f"USGS bağlantı hatası: {e}")

return results
def create_embed(self, earthquake):
"""Discord embed mesajı oluşturur"""
# Renk büyüklüğe göre ayarla
if earthquake['magnitude'] >= 6.0:
color = 0xff0000 # Kırmızı
elif earthquake['magnitude'] >= 5.0:
color = 0xff6600 # Turuncu
else:
color = 0xffcc00 # Sarı

embed = discord.Embed(
title=f" Deprem - {earthquake['magnitude']:.1f} M",
description=earthquake['location'],
color=color,
timestamp=earthquake['time']
)

# Alanlar
embed.add_field(name=" Büyüklük", value=f"{earthquake['magnitude']:.1f}", inline=True)
embed.add_field(name=" Derinlik", value=f"{earthquake['depth']:.1f} km", inline=True)
embed.add_field(name=" Yerel Zaman",
value=earthquake['time'].astimezone().strftime("%d.%m.%Y %H:%M:%S"),
inline=False)
embed.add_field(name=" Kaynak", value=earthquake['source'], inline=True)

# Detay linki (USGS için)
if earthquake.get('detail_url'):
embed.url = earthquake['detail_url']

embed.set_footer(text="Deprem İzleme Botu • Veriler anlık değildir")
return embed
async def on_ready(self):
"""Bot hazır olduğunda çalışır"""
self.logger.info(f'Bot {self.user} olarak giriş yapıldı')
self.logger.info(f'{len(self.guilds)} sunucuda aktif')

# Bot durumunu ayarla
await self.change_presence(
activity=discord.Activity(
type=discord.ActivityType.watching,
name=f"{MIN_MAGNITUDE}+ şiddetinde depremler"
)
)
async def on_disconnect(self):
"""Bağlantı kesildiğinde çalışır"""
global last_checked_time
if last_checked_time:
try:
with open('last_checked.txt', 'w') as f:
f.write(last_checked_time.isoformat())
self.logger.info("Son kontrol zamanı kaydedildi")
except Exception as e:
self.logger.error(f"Son kontrol zamanı kaydedilemedi: {e}")
# Başlangıç
if name == "main":
import asyncio

# Son kontrol zamanını yükle
try:
with open('last_checked.txt', 'r') as f:
last_checked_time = datetime.fromisoformat(f.read().strip())
logging.info(f"Son kontrol: {last_checked_time}")
except (FileNotFoundError, ValueError):
last_checked_time = None
logging.warning("Son kontrol zamanı bulunamadı")

# Botu başlat
bot = DepremBot(intents=intents, ssl=ssl_context)

try:
logging.info("Bot başlatılıyor...")
bot.run(DISCORD_TOKEN)
except discord.LoginFailure:
logging.error("Hata: Geçersiz bot tokeni!")
except KeyboardInterrupt:
logging.info("Bot kapatılıyor...")
except Exception as e:
logging.error(f"Beklenmeyen hata: {type(e).name}: {e}")
TheAny dedi:
Lutfen mesaji duzenleyip tekrar atin. Atarken bu sefer sunun secili olmadigindan emin olun ve duzelt tusuna basmadiginizdam emin olun;
Eki Görüntüle 2477684
Genişletmek için tıkla...
 
Datayı nerden çekiyor bu bot hocam
 
Konuyu cevapla
Menü
Giriş yap
Kaydol
Bu siteyi kullanmak için çerezler gereklidir. Siteyi kullanmaya devam etmek için çerezleri kabul etmelisiniz. Daha Fazlasını Öğren.…