bot/WeatherTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup fileservice and dataservice

Browse Source
This commit is contained in:
Guntis Smaukstelis
2023-12-17 22:48:24 +02:00
parent 578fdd778f
commit 2aa6c5f43c
12 changed files with 73 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/app/Main.scala
+3 -6
View File
@@ -2,7 +2,7 @@ package app
import cats.effect._
import cats.implicits.catsSyntaxTuple2Parallel
import db.{DBConnection, DataService, FileService, PostgresService}
import db.{DBConnection, PostgresService}
import fetch.{FetchService, FileFetchScheduler}
import server.Server
@@ -15,15 +15,12 @@ object Main extends IOApp {
transactor <- DBConnection.transactor[IO]
postgresService <- PostgresService.of(transactor)
_ <- postgresService.createWeatherTable // create table if it does not exists
fileService <- FileService.of
dataService <- DataService.of(fileService, postgresService)
// dataService <- DataService.of(fileService)
fetch <- FetchService.of
fileFetchScheduler <- FileFetchScheduler.of(dataService, fetch)
fileFetchScheduler <- FileFetchScheduler.of(postgresService, fetch)
schedulerTask = fileFetchScheduler.run.compile.drain
server <- Server.of(dataService, fetch)
server <- Server.of(postgresService, fetch)
serverTask = server.run
exitCode <- (serverTask, schedulerTask).parMapN((_, _) => ExitCode.Success)
src/main/scala/db/DBConnection.scala
-1
View File
@@ -20,7 +20,6 @@ object DBConnection {
val raw = URI.create(url)
val name = raw.getPath.substring(1)
println("")
val dbUrl = s"jdbc:postgresql://${raw.getHost}:${raw.getPort}${raw.getPath}?${raw.getQuery}"
val username = raw.getUserInfo.split(":")(0)
val password = raw.getUserInfo.split(":")(1)
src/main/scala/db/DataService.scala
-108
View File
@@ -1,108 +0,0 @@
package db
import cats.effect.unsafe.implicits.global
import cats.effect.{Clock, IO, Ref}
import cats.implicits.toTraverseOps
import fetch.FileNameService
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
import parse.Aggregate
import parse.Aggregate.UserQuery
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
trait DataServiceTrait {
def save(fileName: String, content: String): IO[String]
def readFile(fileName: String): IO[List[String]]
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]]
def getDates: IO[List[LocalDate]]
def getDateFileNames(date: LocalDate): IO[List[String]]
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]]
}
object DataService {
def of(
fileService: FileService,
postgresService: PostgresService
): IO[DataService] = {
for {
log <- Slf4jLogger.create[IO]
// fileNameService = new FileNameService()
// fileNames <- fileNameService.generateLast24Hours
// contents <- fileNames.traverse(fileName => fileService.readFile(fileName)
// .map(content => (fileName, content)))
// state = contents.toMap
// stateRef <- Ref.of[IO, Map[String, List[String]]](state)
} yield new DataService(fileService, postgresService, log)
// } yield new DataService(fileService, postgresService, new FileNameService(), log)
// } yield new DataService(fileService, postgresService, new FileNameService(), log, stateRef)
// } yield new DataService(fileService, new FileNameService(), log, stateRef)
}
}
class DataService private(
fileService: FileService,
postgresService: PostgresService,
// fileNameService: FileNameService,
log: Logger[IO],
// private val state: Ref[IO, Map[String, List[String]]]
) extends DataServiceTrait {
// private def logState: IO[Unit] = {
// state.get.flatMap(currentState => log.info(s"State keys: ${currentState.keys.size}"))
// }
//
// private def filterState: IO[Unit] = {
// fileNameService.generateLast24Hours.flatMap { last24Hours =>
// state.update(st => st.filterKeys(last24Hours.contains).toMap)
// }
// }
def save(fileName: String, content: String): IO[String] = {
// TODO replace unsafeRunSync to redeemWith
// postgresService.save(fileName, content).unsafeRunSync()
// postgresService.save(fileName, content)
for {
result <- fileService.save(fileName, content)
_ <- postgresService.save(fileName, content)
} yield result
// TODO delete this
// fileService.save(fileName, content)
// fileService.save(fileName, content).redeemWith(
// error => IO.raiseError(error),
// savedFileName => {
// state.update(st => st.updated(savedFileName, content.split("\n").toList)) *>
// filterState *>
// logState.as(savedFileName)
// }
// ).onError(error => log.info(s"errr... $error"))
}
def readFile(fileName: String): IO[List[String]] = fileService.readFile(fileName)
def getDateTimeEntries(dateTime: LocalDateTime): IO[List[String]] = postgresService.getDateTimeEntries(dateTime)
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = fileService.getInRange(from, to)
def query(userQuery: UserQuery): IO[Map[String, Option[Aggregate.AggregateValue]]] = postgresService.query(userQuery)
def getDates: IO[List[LocalDate]] = fileService.getDates
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = {
// fileService.getDatesByMonths(monthList)
postgresService.getDatesByMonths(monthList)
}
def getDateFileNames(date: LocalDate): IO[List[String]] = {
// fileService.getDateFileNames(date)
postgresService.getDateFileNames(date)
}
// TODO implement getting full data from state
// def getLast24Hours: IO[List[String]] = {
// state.get.map(_.keys.toList.sorted)
// }
}
src/main/scala/db/FileService.scala
-103
View File
@@ -1,103 +0,0 @@
package db
import cats.effect.{IO, Resource}
import cats.implicits.toTraverseOps
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
import java.io.File
import java.nio.file.{Files, Paths}
import java.time.{LocalDate, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext
import scala.io.Source
import scala.util.Try
object FileService {
def of: IO[FileService] = {
Slf4jLogger.create[IO].map(logger => new FileService(logger))
}
}
class FileService(log: Logger[IO]) extends DataServiceTrait {
private val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
private val dataPath = "./data"
private val nonDuplicatedLines = 34 // takes only first 34 lines of data as rest after 'Zosēni' is duplicated
private def readFileNames(path: String): IO[List[String]] =
IO.blocking(new File(path).listFiles.toList.map(_.getName))
.handleError(_ => List.empty)
private def inRange(fileName: String, from: LocalDateTime, to: LocalDateTime): Boolean = {
def fileToDateTime(fileName: String): Option[LocalDateTime] = {
val dateString = fileName.split("\\.").head
Try(LocalDateTime.parse(dateString, dateFormatter)).toOption
}
val fileDateTime = fileToDateTime(fileName.stripSuffix (".csv"))
fileDateTime match {
case Some (date) => date.plusSeconds (1).isAfter (from) && date.minusSeconds (1).isBefore (to)
case None => false
}
}
def save(fileName: String, content: String): IO[String] = {
val path = Paths.get(s"$dataPath/$fileName")
IO(Files.writeString(path, content))
.attempt
.flatMap {
case Left(error) => log.error(s"Write file '$fileName' failed with error: ${error.getMessage}") *> IO.raiseError(error)
case Right(_) => log.info(s"write: $fileName").as(fileName)
}
}
def readFile(fileName: String): IO[List[String]] = {
val file = new File(dataPath, fileName)
val sourceResource = Resource.fromAutoCloseable(IO.blocking(Source.fromFile(file)))
sourceResource
.use(source => IO.blocking(source.getLines().take(nonDuplicatedLines).toList))
.handleError(_ => List.empty)
}
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = {
for {
fileNames <- readFileNames(dataPath)
.map (_.filter (inRange (_, from, to)))
fileLines <- fileNames.traverse(readFile)
} yield fileLines.flatten
}
// dates in which we have saved data
def getDates: IO[List[LocalDate]] = {
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd")
for {
fileNames <- readFileNames(dataPath)
datesStr <- IO(fileNames.map(_.take(8)).distinct) // take yyyyMMdd
dates <- datesStr.traverse { str =>
IO(LocalDate.parse(str, formatter)).option
}.map(_.flatten)
} yield dates.sorted
}
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = {
val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd")
val monthFormatter = DateTimeFormatter.ofPattern("yyyyMM")
val monthStrList = monthList.map(_.format(monthFormatter))
for {
fileNames <- readFileNames(dataPath)
datesStr <- IO(fileNames.map(_.take(8)).distinct) // take yyyyMMdd
filteredDatesStr = datesStr.filter(date => monthStrList.contains(date.take(6)))
dates <- filteredDatesStr.traverse { str =>
IO(LocalDate.parse(str, dateFormatter)).option
}.map(_.flatten)
} yield dates.sorted
}
def getDateFileNames(date: LocalDate): IO[List[String]] = {
val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMdd")
val dateStr: String = date.format(formatter)
readFileNames(dataPath).map(_.filter(_.startsWith(dateStr)).sorted)
}
}
src/main/scala/db/PostgresService.scala
+1 -3
View File
@@ -22,7 +22,7 @@ object PostgresService {
}
}
class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataServiceTrait {
class PostgresService(transactor: Transactor[IO], log: Logger[IO]) {
private val rigaZone = ZoneId.of("Europe/Riga")
// in pgAdmin run: ```SET TIMEZONE = 'Europe/Riga';```
@@ -40,7 +40,6 @@ class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataS
def query(userQuery: UserQuery): IO[Map[String, Option[AggregateValue]]] = {
if (userQuery.field == "phenomena") { // this handles strings
// TODO query list and distinct values from phenomena
println("EMPTY RESULT!!!!!111")
IO(Map()) // empty result
} else if (List(
AggregateKey.Max,
@@ -145,7 +144,6 @@ class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataS
}
} else {
println("EMPTY RESULT!!!!!")
IO(Map()) // empty result
}
}
src/main/scala/fetch/FileFetchScheduler.scala
+5 -5
View File
@@ -1,23 +1,23 @@
package fetch
import cats.effect.IO
import db.DataServiceTrait
import db.PostgresService
import fs2.Stream
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
object FileFetchScheduler {
def of(dataService: DataServiceTrait, fetch: FetchService): IO[FileFetchScheduler] = {
def of(postgresService: PostgresService, fetch: FetchService): IO[FileFetchScheduler] = {
Scheduler.of.flatMap { scheduler =>
Slf4jLogger.create[IO].map {
new FileFetchScheduler(dataService, fetch, new FileNameService(), scheduler, _)
new FileFetchScheduler(postgresService, fetch, new FileNameService(), scheduler, _)
}
}
}
}
class FileFetchScheduler(dataService: DataServiceTrait, fetch: FetchService, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) {
class FileFetchScheduler(postgresService: PostgresService, fetch: FetchService, fileNameService: FileNameService, scheduler: Scheduler, log: Logger[IO]) {
def run: Stream[IO, Unit] = {
val fetchTask = fileNameService.generateCurrentHour.flatMap(fetch.fetchSingleFile)
scheduler.scheduleTask(fetchTask)
@@ -25,7 +25,7 @@ class FileFetchScheduler(dataService: DataServiceTrait, fetch: FetchService, fil
case Left(fetchErr) =>
log.error(s"Fetch error: $fetchErr")
case Right((name, content)) =>
dataService.save(name, content).attempt.flatMap {
postgresService.save(name, content).attempt.flatMap {
case Left(err) => log.error(s"error: $err")
case Right(savedName) => log.info(s"saved: $savedName")
}
src/main/scala/parse/Main.scala
+8 -8
View File
@@ -2,7 +2,7 @@ package parse
import cats.data.NonEmptyList
import cats.effect.unsafe.implicits.global
import db.FileService
//import db.FileService
import io.circe.syntax.EncoderOps
import parse.Aggregate.{AggregateKey, UserQuery}
@@ -41,14 +41,14 @@ object Main {
// println(s"$key -> $value")
// }
val from = LocalDateTime.parse("20230627_0000", formatter)
val to = LocalDateTime.parse("20230627_2359", formatter)
// val from = LocalDateTime.parse("20230627_0000", formatter)
// val to = LocalDateTime.parse("20230627_2359", formatter)
val fileService = FileService.of.unsafeRunSync()
val lines = fileService.getInRange(from, to).unsafeRunSync()
val query = UserQuery(NonEmptyList.of("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.HOURS, from, to)
val parsed = Parser.queryData(query, lines)
println(parsed.asJson)
// val fileService = FileService.of.unsafeRunSync()
// val lines = fileService.getInRange(from, to).unsafeRunSync()
// val query = UserQuery(NonEmptyList.of("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.HOURS, from, to)
// val parsed = Parser.queryData(query, lines)
// println(parsed.asJson)
// val query2 = UserQuery(List("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.DAYS)
src/main/scala/server/Server.scala
+10 -16
View File
@@ -3,9 +3,9 @@ package server
import cats.effect._
import cats.implicits.toTraverseOps
import com.comcast.ip4s.IpLiteralSyntax
import db.DataService
import db.PostgresService
import fetch.FetchService
import parse.{Aggregate, Parser, WeatherData}
import parse.{Aggregate}
import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, Granularity, ValidateDate, ValidateDateTime, ValidateMonths}
import io.circe.{Json, Printer}
import org.http4s._
@@ -25,19 +25,18 @@ import org.http4s.circe.jsonEncoder
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
import java.time.temporal.ChronoUnit
import scala.concurrent.duration.DurationInt
object Server {
def of(dataService: DataService, fetch: FetchService): IO[Server] = {
def of(postgresService: PostgresService, fetch: FetchService): IO[Server] = {
Slf4jLogger.create[IO].map {
new Server(dataService, fetch, _)
new Server(postgresService, fetch, _)
}
}
}
class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) {
class Server(postgresService: PostgresService, fetch: FetchService, log: Logger[IO]) {
// Define the extension method `pretty` for Json
implicit class JsonPrettyPrinter(json: Json) {
@@ -55,12 +54,7 @@ class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) {
case GET -> Root / "query" / DateTimeRange(from, to) / Granularity(granularity) / CityList(cities) / field / AggKey(key) =>
val userQuery = UserQuery(cities, field, key, granularity, from, to)
// dataService.getInRange(from, to)
// .map(Parser.queryData(userQuery, _))
// .map(result => ResponseWrapper(result, userQuery))
// .flatMap(responseWrapper => Ok(responseWrapper.asJson.pretty))
dataService.query(userQuery)
postgresService.query(userQuery)
.map(result => ResponseWrapper(result, userQuery))
.flatMap(responseWrapper => Ok(responseWrapper.asJson.pretty))
@@ -72,7 +66,7 @@ class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) {
fetchResult = fetchResultEither.getOrElse(List.empty)
(fetchErrors, successDownloads) = fetchResult.partitionMap(identity)
_ <- log.info(s"FETCHED SUCCESSFULLY files: ${successDownloads.size}")
saveResults <- successDownloads.traverse { case (name, content) => dataService.save(name, content).attempt }
saveResults <- successDownloads.traverse { case (name, content) => postgresService.save(name, content).attempt }
(saveErrors, successSaves) = saveResults.partitionMap(identity)
// successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s")
successes = successSaves
@@ -90,19 +84,19 @@ class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) {
// http://0.0.0.0:8080/api/show/months/202304,202305,202306
case GET -> Root / "show" / "months" / ValidateMonths(monthList) =>
dataService.getDatesByMonths(monthList).flatMap(dates =>
postgresService.getDatesByMonths(monthList).flatMap(dates =>
Ok(dates.asJson.pretty)
)
// http://0.0.0.0:8080/api/show/date/20230423
case GET -> Root / "show" / "date" / ValidateDate(date) =>
dataService.getDateFileNames(date).flatMap(fileNames =>
postgresService.getDateFileNames(date).flatMap(fileNames =>
Ok(fileNames.asJson.pretty)
)
// http://0.0.0.0:8080/api/show/datetime/20230423_1300
case GET -> Root / "show" / "datetime" / ValidateDateTime(datetime) =>
dataService.getDateTimeEntries(datetime).flatMap(content => Ok(content.asJson))
postgresService.getDateTimeEntries(datetime).flatMap(content => Ok(content.asJson))
}
private val corsConfig = CORSConfig.default